From 7b306e819039c1f95589d233692038fbbab18fbc Mon Sep 17 00:00:00 2001
From: Anthony Kveder <anthony.kveder@parity.io>
Date: Wed, 17 Jun 2026 07:40:41 +0000
Subject: [PATCH 1/6] Added content-defined chunking

---
 Cargo.lock                              |   9 ++
 Cargo.toml                              |   3 +
 client/src/storage_user.rs              |   7 +-
 docs/design/content-defined-chunking.md |  57 +++++++
 primitives/Cargo.toml                   |   6 +-
 primitives/src/chunking.rs              | 200 ++++++++++++++++++++++++
 primitives/src/lib.rs                   |   3 +
 provider-node/Cargo.toml                |   1 +
 provider-node/src/api.rs                |  39 +++++
 provider-node/src/fs_api.rs             |   6 +-
 provider-node/src/s3_api.rs             |   6 +-
 provider-node/tests/s3_integration.rs   |  95 ++++++++++-
 12 files changed, 415 insertions(+), 17 deletions(-)
 create mode 100644 docs/design/content-defined-chunking.md
 create mode 100644 primitives/src/chunking.rs

diff --git a/Cargo.lock b/Cargo.lock
index 2fa48957..0df39cc8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3117,6 +3117,12 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
 
+[[package]]
+name = "fastcdc"
+version = "3.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf51ceb43e96afbfe4dd5c6f6082af5dfd60e220820b8123792d61963f2ce6bc"
+
 [[package]]
 name = "fastrand"
 version = "2.3.0"
@@ -9689,7 +9695,9 @@ dependencies = [
 name = "storage-primitives"
 version = "0.1.0"
 dependencies = [
+ "fastcdc",
  "parity-scale-codec",
+ "rand 0.8.5",
  "scale-info",
  "serde",
  "sp-core",
@@ -9707,6 +9715,7 @@ dependencies = [
  "hex",
  "parity-scale-codec",
  "parking_lot",
+ "rand 0.8.5",
  "reqwest",
  "rocksdb",
  "serde",
diff --git a/Cargo.toml b/Cargo.toml
index 028a1ae8..4b8a397d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -144,6 +144,9 @@ tower-http = { version = "0.6", features = ["cors", "trace"] }
 # Crypto
 blake2 = { version = "0.10", default-features = false }
 
+# Chunking
+fastcdc = { version = "3.1" }
+
 # Subxt (chain interaction for off-chain clients)
 subxt = { version = "0.44.3" }
 subxt-signer = { version = "0.44.3", features = ["sr25519"] }
diff --git a/client/src/storage_user.rs b/client/src/storage_user.rs
index 7504920e..bde63002 100644
--- a/client/src/storage_user.rs
+++ b/client/src/storage_user.rs
@@ -465,12 +465,9 @@ impl StorageUserClient {
     fn chunk_data(data: &[u8], strategy: ChunkingStrategy) -> Vec<Vec<u8>> {
         match strategy {
             ChunkingStrategy::Fixed(chunk_size) => {
-                data.chunks(chunk_size).map(|c| c.to_vec()).collect()
-            }
-            ChunkingStrategy::ContentDefined => {
-                // TODO: Implement content-defined chunking
-                Self::chunk_data(data, ChunkingStrategy::Fixed(256 * 1024))
+                storage_primitives::chunking::chunk_fixed(data, chunk_size)
             }
+            ChunkingStrategy::ContentDefined => storage_primitives::chunking::chunk_cdc(data),
         }
     }
 
diff --git a/docs/design/content-defined-chunking.md b/docs/design/content-defined-chunking.md
new file mode 100644
index 00000000..64c7bc69
--- /dev/null
+++ b/docs/design/content-defined-chunking.md
@@ -0,0 +1,57 @@
+# Content-Defined Chunking (CDC)
+
+## Why
+
+The provider stores file bytes as content-addressed chunks: each chunk's key is `blake2_256(bytes)`, so two PUTs that produce a byte-identical chunk get stored once. That dedup is automatic — *if* the chunker actually produces identical chunks across edits.
+
+With a fixed-size chunker, it doesn't. Inserting a single byte at the start of a file shifts every downstream byte by one, so every chunk has different bytes, so every chunk has a different hash. v2's PUT effectively re-uploads the entire file.
+
+Content-defined chunking solves this by choosing chunk boundaries from the *content* (via a rolling hash), not from fixed byte offsets. When you insert bytes in the middle of a file, the boundaries downstream of the edit fall at the same content positions as before; the chunks between them are byte-identical to v1's chunks and dedup through content addressing.
+
+## Algorithm
+
+[FastCDC](https://www.usenix.org/system/files/conference/atc16/atc16-paper-xia.pdf), via the [`fastcdc-rs`](https://crates.io/crates/fastcdc) crate. Picked over Rabin fingerprinting for:
+- ~3-5× higher throughput (gear-hash based)
+- Equivalent dedup ratio
+- Maintained Rust implementation in production use by `restic` and others
+
+Mechanically: slide a 32-byte window across the input, compute a gear hash of the window contents at each position, emit a boundary when the low bits of the hash match a fixed pattern (forced boundaries enforce `max_size`; the hash check is skipped below `min_size`).
+
+## Parameters
+
+| Parameter | Value | Why |
+|---|---|---|
+| `CDC_MIN_SIZE` | 64 KiB | Avoid tiny chunks that inflate metadata / proof overhead |
+| `CDC_AVG_SIZE` | 256 KiB | Matches the prior `DEFAULT_CHUNK_SIZE` so MMR leaf counts and proof depths stay comparable |
+| `CDC_MAX_SIZE` | 1 MiB | Bounds worst-case proof and read cost |
+| Window size | 32 bytes (FastCDC default) | |
+
+These live in [`primitives/src/chunking.rs`](../../primitives/src/chunking.rs).
+
+## What changes, what doesn't
+
+**Changed.** Chunks are now variable-size (between 64 KiB and 1 MiB). The provider's S3 (`PUT /s3/.../object`) and FS (`PUT /fs/.../file`) handlers chunk via `chunk_cdc_borrowed`. The Rust client SDK's `ChunkingStrategy::ContentDefined` arm uses the same chunker (previously a TODO that fell back to fixed-256K).
+
+**Unchanged.** The MMR is still leaf-per-chunk and chunk-size-agnostic — each leaf is `blake2_256(chunk_bytes)`. The Merkle proof model is unchanged. The S3/FS GET handlers still reassemble via `collect_chunks(data_root)`, which walks the tree by hash and doesn't assume any chunk size. Content addressing and per-bucket isolation are unchanged.
+
+The `GET /read?data_root=&offset=&length=` byte-range endpoint **still assumes fixed-size** (its arithmetic computes `chunk_index = offset / DEFAULT_CHUNK_SIZE`). It is therefore unsafe to call on CDC-chunked roots. For whole-file historical fetch, use the new `GET /content?data_root=` endpoint, which walks the manifest. Updating `/read` for variable-size byte-range reads is deferred until a caller actually needs it.
+
+## Fixed vs CDC trade-offs
+
+| | Fixed-size | CDC |
+|---|---|---|
+| Implementation | Trivial | FastCDC (~200 LOC dep) |
+| Throughput | Memory-bandwidth-bound | ~1 GB/s on a modern CPU |
+| Chunk-size variance | Zero (last chunk smaller) | Bounded by `[min, max]` |
+| Insert/delete dedup | ❌ Cascades downstream | ✅ Only chunks straddling the edit change |
+| In-place overwrite dedup | ✅ Works | ✅ Works |
+| Append dedup | ✅ Works (only trailing chunk changes) | ✅ Works |
+| Best for | Binary blobs with fixed-offset fields; embedded use where every byte counts | Text, structured data, anything with shifting edits |
+
+`ChunkingStrategy::Fixed(usize)` remains the SDK default; CDC is opt-in via `ChunkingStrategy::ContentDefined`. The provider's HTTP upload endpoints (S3 / FS) always use CDC since their callers don't pick a strategy.
+
+## Verification
+
+- Unit tests in `primitives/src/chunking.rs::tests` cover determinism, size-distribution bounds, byte-equal reassembly, and ≥ 90% chunk reuse on mid-file insertion / deletion of 8 MiB random data.
+- Integration test `test_s3_cdc_dedup_across_versions` in `provider-node/tests/s3_integration.rs` PUTs an 8 MiB blob, then a mid-file-edited variant, and asserts the second PUT adds fewer than `total_nodes(v1) / 4` new nodes.
+- `test_get_content_returns_full_bytes` exercises the new `/content` endpoint.
diff --git a/primitives/Cargo.toml b/primitives/Cargo.toml
index 46e12d65..cde53f7a 100644
--- a/primitives/Cargo.toml
+++ b/primitives/Cargo.toml
@@ -12,8 +12,12 @@ codec = { workspace = true }
 scale-info = { workspace = true }
 sp-core = { workspace = true }
 serde = { workspace = true, optional = true }
+fastcdc = { workspace = true, optional = true }
+
+[dev-dependencies]
+rand = "0.8"
 
 [features]
 default = ["std"]
-std = ["codec/std", "scale-info/std", "serde/std", "sp-core/std"]
+std = ["codec/std", "dep:fastcdc", "scale-info/std", "serde/std", "sp-core/std"]
 serde = ["dep:serde", "sp-core/serde"]
diff --git a/primitives/src/chunking.rs b/primitives/src/chunking.rs
new file mode 100644
index 00000000..b82795cf
--- /dev/null
+++ b/primitives/src/chunking.rs
@@ -0,0 +1,200 @@
+//! Chunking strategies for content storage.
+//!
+//! - [`chunk_fixed`]: even-sized chunks. A single-byte insertion at the start
+//!   of a file invalidates every downstream chunk.
+//! - [`chunk_cdc`]: content-defined chunks via FastCDC. Boundaries align on
+//!   content, so an insertion only changes chunks straddling the edit.
+
+use alloc::vec::Vec;
+
+/// FastCDC minimum chunk size — below this, no boundary is emitted regardless
+/// of the rolling hash. Keeps metadata overhead bounded for tiny files.
+pub const CDC_MIN_SIZE: u32 = 64 * 1024;
+
+/// FastCDC average (target) chunk size. Matches [`DEFAULT_CHUNK_SIZE`] so MMR
+/// leaf counts and proof depths stay comparable to the fixed-size chunker.
+///
+/// [`DEFAULT_CHUNK_SIZE`]: crate::DEFAULT_CHUNK_SIZE
+pub const CDC_AVG_SIZE: u32 = 256 * 1024;
+
+/// FastCDC maximum chunk size — forced boundary even with no hash match.
+/// Bounds worst-case proof / read cost.
+pub const CDC_MAX_SIZE: u32 = 1024 * 1024;
+
+/// Split `data` into fixed-size chunks of `chunk_size` bytes (the last chunk
+/// may be smaller).
+pub fn chunk_fixed(data: &[u8], chunk_size: usize) -> Vec<Vec<u8>> {
+    data.chunks(chunk_size).map(<[u8]>::to_vec).collect()
+}
+
+/// Split `data` into content-defined chunks using FastCDC with the
+/// [`CDC_MIN_SIZE`] / [`CDC_AVG_SIZE`] / [`CDC_MAX_SIZE`] parameters.
+pub fn chunk_cdc(data: &[u8]) -> Vec<Vec<u8>> {
+    chunk_cdc_with(data, CDC_MIN_SIZE, CDC_AVG_SIZE, CDC_MAX_SIZE)
+}
+
+/// Like [`chunk_cdc`] but returns slices of the input. Avoids allocating until
+/// the caller decides what to do with each chunk.
+pub fn chunk_cdc_borrowed(data: &[u8]) -> Vec<&[u8]> {
+    if data.is_empty() {
+        return Vec::new();
+    }
+    fastcdc::v2020::FastCDC::new(data, CDC_MIN_SIZE, CDC_AVG_SIZE, CDC_MAX_SIZE)
+        .map(|chunk| &data[chunk.offset..chunk.offset + chunk.length])
+        .collect()
+}
+
+/// Like [`chunk_cdc`] but with explicit parameters. Exposed for tests and
+/// future tuning; callers should prefer [`chunk_cdc`].
+pub fn chunk_cdc_with(data: &[u8], min: u32, avg: u32, max: u32) -> Vec<Vec<u8>> {
+    if data.is_empty() {
+        return Vec::new();
+    }
+    fastcdc::v2020::FastCDC::new(data, min, avg, max)
+        .map(|chunk| data[chunk.offset..chunk.offset + chunk.length].to_vec())
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::blake2_256;
+    use rand::{RngCore, SeedableRng};
+    use std::collections::HashSet;
+
+    fn random_bytes(seed: u64, len: usize) -> Vec<u8> {
+        let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+        let mut buf = vec![0u8; len];
+        rng.fill_bytes(&mut buf);
+        buf
+    }
+
+    fn chunk_hashes(chunks: &[Vec<u8>]) -> HashSet<[u8; 32]> {
+        chunks.iter().map(|c| blake2_256(c).0).collect()
+    }
+
+    #[test]
+    fn cdc_size_distribution_within_bounds() {
+        let data = random_bytes(1, 4 * 1024 * 1024);
+        let chunks = chunk_cdc(&data);
+        assert!(chunks.len() > 1, "expected multiple chunks for 4 MiB");
+        // Every chunk except the trailing one must respect [min, max].
+        for chunk in chunks.iter().take(chunks.len() - 1) {
+            assert!(
+                chunk.len() >= CDC_MIN_SIZE as usize,
+                "chunk smaller than CDC_MIN_SIZE: {}",
+                chunk.len()
+            );
+            assert!(
+                chunk.len() <= CDC_MAX_SIZE as usize,
+                "chunk larger than CDC_MAX_SIZE: {}",
+                chunk.len()
+            );
+        }
+    }
+
+    #[test]
+    fn cdc_is_deterministic() {
+        let data = random_bytes(2, 1024 * 1024);
+        let a = chunk_cdc(&data);
+        let b = chunk_cdc(&data);
+        assert_eq!(a, b);
+    }
+
+    #[test]
+    fn cdc_reassembles_byte_equal() {
+        let data = random_bytes(3, 1024 * 1024 + 12345);
+        let chunks = chunk_cdc(&data);
+        let mut joined = Vec::with_capacity(data.len());
+        for c in &chunks {
+            joined.extend_from_slice(c);
+        }
+        assert_eq!(joined, data);
+    }
+
+    /// 8 MiB is the smallest file size where avg-256K chunking gives enough
+    /// leaves (~32) that one chunk straddling the edit is a small fraction;
+    /// at 1 MiB the test would cap at 75% regardless of CDC quality.
+    const EDIT_TEST_LEN: usize = 8 * 1024 * 1024;
+
+    /// Insertion in the middle must reuse most chunks. Fixed-size would reuse
+    /// roughly half (only chunks before the insertion); CDC reuses everything
+    /// outside the window straddling the edit.
+    #[test]
+    fn cdc_insertion_reuses_most_chunks() {
+        let v1 = random_bytes(4, EDIT_TEST_LEN);
+        let mut v2 = v1.clone();
+        let insertion_point = v1.len() / 2;
+        v2.splice(
+            insertion_point..insertion_point,
+            random_bytes(5, 200).iter().copied(),
+        );
+
+        let v1_chunks = chunk_cdc(&v1);
+        let v2_chunks = chunk_cdc(&v2);
+        let v1_hashes = chunk_hashes(&v1_chunks);
+        let v2_hashes = chunk_hashes(&v2_chunks);
+        let shared = v1_hashes.intersection(&v2_hashes).count();
+        let ratio = shared as f64 / v1_hashes.len() as f64;
+        assert!(
+            ratio >= 0.9,
+            "expected ≥ 90% chunk reuse on mid-file insertion, got {:.2}% \
+             (v1={}, v2={}, shared={})",
+            ratio * 100.0,
+            v1_hashes.len(),
+            v2_hashes.len(),
+            shared
+        );
+    }
+
+    #[test]
+    fn cdc_deletion_reuses_most_chunks() {
+        let v1 = random_bytes(6, EDIT_TEST_LEN);
+        let mut v2 = v1.clone();
+        let cut = v1.len() / 2;
+        v2.drain(cut..cut + 200);
+
+        let v1_hashes = chunk_hashes(&chunk_cdc(&v1));
+        let v2_hashes = chunk_hashes(&chunk_cdc(&v2));
+        let shared = v1_hashes.intersection(&v2_hashes).count();
+        let ratio = shared as f64 / v1_hashes.len() as f64;
+        assert!(
+            ratio >= 0.9,
+            "expected ≥ 90% chunk reuse on mid-file deletion, got {:.2}%",
+            ratio * 100.0
+        );
+    }
+
+    /// Appending data must leave every prior chunk's CID unchanged except
+    /// possibly the trailing partial chunk.
+    #[test]
+    fn cdc_append_preserves_prior_chunks() {
+        let v1 = random_bytes(7, 1024 * 1024);
+        let mut v2 = v1.clone();
+        v2.extend_from_slice(&random_bytes(8, 50_000));
+
+        let v1_hashes = chunk_hashes(&chunk_cdc(&v1));
+        let v2_hashes = chunk_hashes(&chunk_cdc(&v2));
+        let shared = v1_hashes.intersection(&v2_hashes).count();
+        // The final v1 chunk may have been smaller than CDC_MIN and got
+        // merged with the new tail; allow at most one prior chunk to differ.
+        assert!(
+            shared + 1 >= v1_hashes.len(),
+            "expected all-but-one v1 chunks to survive append, got shared={} of {}",
+            shared,
+            v1_hashes.len()
+        );
+    }
+
+    #[test]
+    fn fixed_round_trips() {
+        let data = random_bytes(9, 1024 * 1024 + 7);
+        let chunks = chunk_fixed(&data, 256 * 1024);
+        // Last chunk smaller, others exactly 256 KiB.
+        for chunk in chunks.iter().take(chunks.len() - 1) {
+            assert_eq!(chunk.len(), 256 * 1024);
+        }
+        let joined: Vec<u8> = chunks.into_iter().flatten().collect();
+        assert_eq!(joined, data);
+    }
+}
diff --git a/primitives/src/lib.rs b/primitives/src/lib.rs
index c521800c..4322266c 100644
--- a/primitives/src/lib.rs
+++ b/primitives/src/lib.rs
@@ -13,6 +13,9 @@ use core::fmt::Debug;
 use scale_info::TypeInfo;
 use sp_core::H256;
 
+#[cfg(feature = "std")]
+pub mod chunking;
+
 /// Bucket ID is a stable, unique identifier (not an index into a collection).
 /// Using u64 ensures IDs never get reused even if buckets are deleted.
 pub type BucketId = u64;
diff --git a/provider-node/Cargo.toml b/provider-node/Cargo.toml
index 4db5e4a4..df349934 100644
--- a/provider-node/Cargo.toml
+++ b/provider-node/Cargo.toml
@@ -33,3 +33,4 @@ codec = { workspace = true, features = ["std"] }
 [dev-dependencies]
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
 serde_json = { workspace = true }
+rand = "0.8"
diff --git a/provider-node/src/api.rs b/provider-node/src/api.rs
index 3fe57edb..6bb00f46 100644
--- a/provider-node/src/api.rs
+++ b/provider-node/src/api.rs
@@ -12,6 +12,8 @@ use crate::types::*;
 use crate::ProviderState;
 use axum::{
     extract::{DefaultBodyLimit, Query, State},
+    http::{header, StatusCode},
+    response::{IntoResponse, Response},
     routing::{get, post, put},
     Json, Router,
 };
@@ -36,6 +38,7 @@ pub fn create_router(state: Arc<ProviderState>) -> Router {
         // Commit and read
         .route("/commit", post(commit))
         .route("/read", get(read_chunks))
+        .route("/content", get(get_content))
         // Commitment and proofs
         .route("/commitment", get(get_commitment))
         .route("/checkpoint-signature", get(get_checkpoint_signature))
@@ -322,6 +325,42 @@ async fn read_chunks(
     Ok(Json(ReadResponse { chunks }))
 }
 
+#[derive(serde::Deserialize)]
+struct GetContentQuery {
+    data_root: String,
+}
+
+/// GET /content?data_root=<hex>
+///
+/// Reassembles bytes by CID, regardless of which S3/FS key currently points
+/// at it. `/read` assumes fixed-size chunks and can't serve CDC-chunked roots.
+async fn get_content(
+    State(state): State<Arc<ProviderState>>,
+    Query(query): Query<GetContentQuery>,
+) -> Result<Response, Error> {
+    let root_bytes = hex_decode(&query.data_root).map_err(|_| Error::InvalidHash {
+        expected: query.data_root.clone(),
+        actual: "invalid hex".to_string(),
+    })?;
+    let data_root = H256::from_slice(&root_bytes);
+
+    let chunks = state.storage.collect_chunks(data_root);
+    if chunks.is_empty() && data_root != H256::zero() {
+        return Err(Error::NodeNotFound(query.data_root));
+    }
+    let total: usize = chunks.iter().map(Vec::len).sum();
+    let mut body = Vec::with_capacity(total);
+    for chunk in chunks {
+        body.extend_from_slice(&chunk);
+    }
+    let mut response = (StatusCode::OK, body).into_response();
+    response.headers_mut().insert(
+        header::CONTENT_TYPE,
+        "application/octet-stream".parse().unwrap(),
+    );
+    Ok(response)
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Commitment and Proofs
 // ─────────────────────────────────────────────────────────────────────────────
diff --git a/provider-node/src/fs_api.rs b/provider-node/src/fs_api.rs
index f8113212..4529a65d 100644
--- a/provider-node/src/fs_api.rs
+++ b/provider-node/src/fs_api.rs
@@ -66,12 +66,12 @@ pub async fn fs_put_file(
     // Initialize bucket if needed
     let _ = state.storage.init_bucket(bucket_id, u64::MAX);
 
-    // 1. Split into chunks (256 KiB)
-    let chunk_size = storage_primitives::DEFAULT_CHUNK_SIZE as usize;
+    // An empty file still gets one empty leaf so its data_root is the
+    // deterministic blake2_256(&[]).
     let chunks: Vec<&[u8]> = if data.is_empty() {
         vec![&[]]
     } else {
-        data.chunks(chunk_size).collect()
+        storage_primitives::chunking::chunk_cdc_borrowed(&data)
     };
 
     // 2. Hash and store each chunk
diff --git a/provider-node/src/s3_api.rs b/provider-node/src/s3_api.rs
index 84eaf315..984bfd40 100644
--- a/provider-node/src/s3_api.rs
+++ b/provider-node/src/s3_api.rs
@@ -53,12 +53,12 @@ pub async fn s3_put_object(
     // Initialize bucket if needed
     let _ = state.storage.init_bucket(bucket_id, u64::MAX);
 
-    // 1. Split into chunks (256 KiB)
-    let chunk_size = storage_primitives::DEFAULT_CHUNK_SIZE as usize;
+    // An empty object still gets one empty leaf so its data_root is the
+    // deterministic blake2_256(&[]).
     let chunks: Vec<&[u8]> = if data.is_empty() {
         vec![&[]]
     } else {
-        data.chunks(chunk_size).collect()
+        storage_primitives::chunking::chunk_cdc_borrowed(&data)
     };
 
     // 2. Hash and store each chunk
diff --git a/provider-node/tests/s3_integration.rs b/provider-node/tests/s3_integration.rs
index c6886592..47936b8b 100644
--- a/provider-node/tests/s3_integration.rs
+++ b/provider-node/tests/s3_integration.rs
@@ -316,9 +316,8 @@ async fn test_s3_index_root() {
 async fn test_s3_large_file_multi_chunk() {
     let server = TestServer::new().await;
 
-    // Create data larger than one chunk (256 KiB = 262144 bytes)
-    // Use 300 KB to get 2 chunks
-    let data = vec![0x42u8; 300 * 1024];
+    // Big enough that CDC will emit multiple chunks (>= CDC_MAX_SIZE = 1 MiB).
+    let data = vec![0x42u8; 3 * 1024 * 1024];
 
     let response = server
         .client
@@ -331,7 +330,7 @@ async fn test_s3_large_file_multi_chunk() {
 
     assert_eq!(response.status(), StatusCode::OK);
     let body: Value = response.json().await.unwrap();
-    assert_eq!(body["size"], 300 * 1024);
+    assert_eq!(body["size"], 3 * 1024 * 1024);
 
     // GET and verify round-trip
     let response = server
@@ -343,7 +342,7 @@ async fn test_s3_large_file_multi_chunk() {
 
     assert_eq!(response.status(), StatusCode::OK);
     let returned_data = response.bytes().await.unwrap();
-    assert_eq!(returned_data.len(), 300 * 1024);
+    assert_eq!(returned_data.len(), 3 * 1024 * 1024);
     assert_eq!(returned_data.as_ref(), data.as_slice());
 }
 
@@ -424,3 +423,89 @@ async fn test_s3_mmr_proof_works_with_s3_data() {
     assert!(body["leaf"]["data_root"].is_string());
     assert!(body["proof"]["peaks"].is_array());
 }
+
+#[tokio::test]
+async fn test_s3_cdc_dedup_across_versions() {
+    use rand::{RngCore, SeedableRng};
+
+    let server = TestServer::new().await;
+
+    let mut rng = rand::rngs::StdRng::seed_from_u64(0xCD0);
+    let mut v1 = vec![0u8; 8 * 1024 * 1024];
+    rng.fill_bytes(&mut v1);
+    let mut v2 = v1.clone();
+    let mid = v2.len() / 2;
+    v2.splice(mid..mid, [0xAB; 200].iter().copied());
+
+    async fn total_nodes(server: &TestServer) -> u64 {
+        let r: Value = server
+            .client
+            .get(server.url("/stats"))
+            .send()
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        r["total_nodes"].as_u64().unwrap()
+    }
+
+    let r = server
+        .client
+        .put(server.url("/s3/1/object?key=doc.bin"))
+        .body(v1.clone())
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status(), StatusCode::OK);
+    let after_v1 = total_nodes(&server).await;
+    assert!(after_v1 > 0);
+
+    let r = server
+        .client
+        .put(server.url("/s3/1/object?key=doc.bin"))
+        .body(v2.clone())
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status(), StatusCode::OK);
+    let after_v2 = total_nodes(&server).await;
+
+    // A fixed-size chunker would invalidate every chunk past the insertion;
+    // CDC should stay well under v1/4.
+    let added = after_v2 - after_v1;
+    assert!(
+        added < after_v1 / 4,
+        "CDC dedup weak: v1 added {after_v1} nodes, v2 added {added} more \
+         (expected < {} ≈ v1/4)",
+        after_v1 / 4
+    );
+}
+
+#[tokio::test]
+async fn test_get_content_returns_full_bytes() {
+    let server = TestServer::new().await;
+
+    let body = vec![0xA5u8; 2 * 1024 * 1024];
+    let put: Value = server
+        .client
+        .put(server.url("/s3/1/object?key=blob.bin"))
+        .body(body.clone())
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let data_root = put["data_root"].as_str().unwrap().to_string();
+
+    let r = server
+        .client
+        .get(server.url(&format!("/content?data_root={data_root}")))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status(), StatusCode::OK);
+    let returned = r.bytes().await.unwrap();
+    assert_eq!(returned.as_ref(), body.as_slice());
+}

From 9e985b529e6d45e2bb0ded34c1a460d27afb470e Mon Sep 17 00:00:00 2001
From: Anthony Kveder <anthony.kveder@parity.io>
Date: Wed, 24 Jun 2026 10:41:23 +0000
Subject: [PATCH 2/6] Address CDC review: workspace rand dep, SPDX header, no
 unwrap in /content

---
 CLAUDE.md                  | 7 +++++++
 Cargo.toml                 | 1 +
 primitives/Cargo.toml      | 2 +-
 primitives/src/chunking.rs | 2 ++
 provider-node/Cargo.toml   | 2 +-
 provider-node/src/api.rs   | 4 ++--
 6 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/CLAUDE.md b/CLAUDE.md
index cc100aaf..1e58c5f9 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -15,6 +15,13 @@
 - ALWAYS run `/format` before creating any git commit
 - This ensures all code follows project formatting standards (Rust, TOML, feature propagation) and passes clippy
 
+**File headers:**
+- Every new source file (Rust, TypeScript, JavaScript, Solidity) MUST start with an SPDX license header on the first line, matching the surrounding crate / package. Check a sibling file in the same crate to copy the exact identifier (`Apache-2.0`, `GPL-3.0-only`, etc.).
+- Examples: `// SPDX-License-Identifier: Apache-2.0` for the `primitives` crate, `// SPDX-License-Identifier: GPL-3.0-only` for `provider-node` and the `user-interfaces/` workspace.
+
+**Dependencies:**
+- New external dependencies go in the workspace `Cargo.toml` under `[workspace.dependencies]`, then each consuming crate references them via `{ workspace = true }`. Don't inline version literals in per-crate manifests; that includes `[dev-dependencies]`.
+
 ## Project Overview
 
 Scalable Web3 Storage is a decentralized storage system built on Substrate with game-theoretic guarantees. Storage providers lock stake and face slashing for data loss, while the chain acts as a credible threat rather than the hot path.
diff --git a/Cargo.toml b/Cargo.toml
index a9b59a9f..02ef3c14 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -154,6 +154,7 @@ subxt = { version = "0.44.3" }
 subxt-signer = { version = "0.44.3", features = ["sr25519"] }
 
 # Testing
+rand = "0.8"
 sp-keyring = { version = "46.0.0", default-features = false }
 sp-keystore = { version = "0.46.0", default-features = false }
 
diff --git a/primitives/Cargo.toml b/primitives/Cargo.toml
index c32d9b02..800c428f 100644
--- a/primitives/Cargo.toml
+++ b/primitives/Cargo.toml
@@ -15,7 +15,7 @@ serde = { workspace = true, optional = true }
 fastcdc = { workspace = true, optional = true }
 
 [dev-dependencies]
-rand = "0.8"
+rand = { workspace = true }
 
 [features]
 default = ["std"]
diff --git a/primitives/src/chunking.rs b/primitives/src/chunking.rs
index b82795cf..bf335692 100644
--- a/primitives/src/chunking.rs
+++ b/primitives/src/chunking.rs
@@ -1,3 +1,5 @@
+// SPDX-License-Identifier: Apache-2.0
+
 //! Chunking strategies for content storage.
 //!
 //! - [`chunk_fixed`]: even-sized chunks. A single-byte insertion at the start
diff --git a/provider-node/Cargo.toml b/provider-node/Cargo.toml
index 3a13dcc2..e15c67eb 100644
--- a/provider-node/Cargo.toml
+++ b/provider-node/Cargo.toml
@@ -37,5 +37,5 @@ async-trait = "0.1"
 [dev-dependencies]
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util", "time"] }
 serde_json = { workspace = true }
-rand = "0.8"
+rand = { workspace = true }
 tempfile = "3"
diff --git a/provider-node/src/api.rs b/provider-node/src/api.rs
index ea7b80b6..6c030637 100644
--- a/provider-node/src/api.rs
+++ b/provider-node/src/api.rs
@@ -15,7 +15,7 @@ use crate::types::*;
 use crate::ProviderState;
 use axum::{
     extract::{ConnectInfo, DefaultBodyLimit, Query, Request, State},
-    http::{header, StatusCode},
+    http::{header, HeaderValue, StatusCode},
     middleware::{from_fn_with_state, Next},
     response::{IntoResponse, Response},
     routing::{get, post, put},
@@ -419,7 +419,7 @@ async fn get_content(
     let mut response = (StatusCode::OK, body).into_response();
     response.headers_mut().insert(
         header::CONTENT_TYPE,
-        "application/octet-stream".parse().unwrap(),
+        HeaderValue::from_static("application/octet-stream"),
     );
     Ok(response)
 }

From ce2e6c62d8ed9326a757aa398bd99e0516f76337 Mon Sep 17 00:00:00 2001
From: Anthony Kveder <anthony.kveder@parity.io>
Date: Wed, 24 Jun 2026 11:05:28 +0000
Subject: [PATCH 3/6] Added guard for rejecting non-fixed roots in /read,
 updated CDC .md

---
 docs/design/content-defined-chunking.md |  8 ++++-
 provider-node/src/api.rs                | 45 ++++++++++++++++++++++++-
 provider-node/src/error.rs              | 19 +++++++++++
 provider-node/tests/s3_integration.rs   | 37 ++++++++++++++++++++
 4 files changed, 107 insertions(+), 2 deletions(-)

diff --git a/docs/design/content-defined-chunking.md b/docs/design/content-defined-chunking.md
index 64c7bc69..5f972bbc 100644
--- a/docs/design/content-defined-chunking.md
+++ b/docs/design/content-defined-chunking.md
@@ -34,7 +34,13 @@ These live in [`primitives/src/chunking.rs`](../../primitives/src/chunking.rs).
 
 **Unchanged.** The MMR is still leaf-per-chunk and chunk-size-agnostic — each leaf is `blake2_256(chunk_bytes)`. The Merkle proof model is unchanged. The S3/FS GET handlers still reassemble via `collect_chunks(data_root)`, which walks the tree by hash and doesn't assume any chunk size. Content addressing and per-bucket isolation are unchanged.
 
-The `GET /read?data_root=&offset=&length=` byte-range endpoint **still assumes fixed-size** (its arithmetic computes `chunk_index = offset / DEFAULT_CHUNK_SIZE`). It is therefore unsafe to call on CDC-chunked roots. For whole-file historical fetch, use the new `GET /content?data_root=` endpoint, which walks the manifest. Updating `/read` for variable-size byte-range reads is deferred until a caller actually needs it.
+The `GET /read?data_root=&offset=&length=` byte-range endpoint assumes fixed-size chunks — its arithmetic is `chunk_index = offset / DEFAULT_CHUNK_SIZE`. Now that S3 and FS uploads always use CDC, that's wrong for the **common** case rather than a rare one, and `get_chunk_at_index` would still hand back *a* chunk, so the endpoint would silently serve misaligned bytes with a 200.
+
+To prevent that, `/read` now **guards** every call: it walks the leaves under `data_root` and rejects with `422 variable_chunk_root` if any non-trailing leaf isn't exactly `DEFAULT_CHUNK_SIZE`. Implementation: `data_root_is_fixed_size` in `provider-node/src/api.rs`; integration coverage in `test_read_rejects_cdc_root`. For whole-file historical fetch, use the new `GET /content?data_root=` endpoint, which walks the manifest and works for any chunking strategy.
+
+Worth flagging: **roots don't record their chunking strategy.** A `data_root` is just a Merkle root over a list of `blake2_256(chunk)` leaves; nothing in the tree distinguishes a fixed-256K-chunked root from a CDC root from anything else. Any future endpoint that does offset/length arithmetic is therefore globally unsafe until we either (a) tag roots with their strategy (e.g. an extra storage map alongside the bucket's S3/FS index), (b) make the read side fully strategy-agnostic by walking a manifest with explicit per-chunk offsets, or (c) only emit one canonical chunking. Right now the guard substitutes for that — by content-inspecting the leaves at request time — and is intentionally conservative (rejects empty roots, rejects anything where any non-trailing leaf differs from `DEFAULT_CHUNK_SIZE`).
+
+A real variable-size byte-range implementation of `/read` is deferred until a caller needs it.
 
 ## Fixed vs CDC trade-offs
 
diff --git a/provider-node/src/api.rs b/provider-node/src/api.rs
index 6c030637..8c204b8b 100644
--- a/provider-node/src/api.rs
+++ b/provider-node/src/api.rs
@@ -359,7 +359,15 @@ async fn read_chunks(
     })?;
     let data_root = H256::from_slice(&root_bytes);
 
-    // Calculate chunk indices
+    // `/read`'s offset/length math assumes every leaf except the last is exactly
+    // `DEFAULT_CHUNK_SIZE` bytes. CDC-chunked roots break that assumption — the
+    // math would still compute *a* chunk_index but the bytes wouldn't line up,
+    // silently serving misaligned data. Reject those roots; whole-file readers
+    // should use `GET /content?data_root=` which walks the manifest.
+    if !data_root_is_fixed_size(&*state.storage, data_root)? {
+        return Err(Error::VariableChunkRoot);
+    }
+
     let chunk_size = storage_primitives::DEFAULT_CHUNK_SIZE as u64;
     let start_chunk = query.offset / chunk_size;
     let end_chunk = (query.offset + query.length).div_ceil(chunk_size);
@@ -388,6 +396,41 @@ async fn read_chunks(
     Ok(Json(ReadResponse { chunks }))
 }
 
+/// Walks the leaves under `data_root` and verifies all-but-last are exactly
+/// `DEFAULT_CHUNK_SIZE` bytes. Used by `/read` to reject CDC roots — the
+/// fixed-size offset math would otherwise serve misaligned bytes.
+///
+/// Returns `Ok(false)` for an empty / unknown root (no leaves to validate);
+/// downstream `get_chunk_at_index` already errors cleanly on those.
+fn data_root_is_fixed_size(
+    storage: &dyn crate::storage::StorageBackend,
+    data_root: H256,
+) -> Result<bool, Error> {
+    let hashes = storage.collect_chunk_hashes(data_root);
+    if hashes.is_empty() {
+        return Ok(false);
+    }
+    let chunk_size = storage_primitives::DEFAULT_CHUNK_SIZE as usize;
+    let (last, prefix) = hashes
+        .split_last()
+        .expect("hashes is non-empty (checked above)");
+    for hash in prefix {
+        let Some(node) = storage.get_node(hash) else {
+            return Err(Error::NodeNotFound(format!(
+                "0x{}",
+                hex_encode(hash.as_bytes())
+            )));
+        };
+        if node.data.len() != chunk_size {
+            return Ok(false);
+        }
+    }
+    // The trailing leaf may be smaller (last fixed-size chunk often is); reject
+    // only if it's *larger*, which a fixed-size chunker can't produce.
+    let last_len = storage.get_node(last).map_or(0, |n| n.data.len());
+    Ok(last_len <= chunk_size)
+}
+
 #[derive(serde::Deserialize)]
 struct GetContentQuery {
     data_root: String,
diff --git a/provider-node/src/error.rs b/provider-node/src/error.rs
index cfb61674..3c349e15 100644
--- a/provider-node/src/error.rs
+++ b/provider-node/src/error.rs
@@ -105,6 +105,13 @@ pub enum Error {
 
     #[error("Too many requests")]
     RateLimited,
+
+    /// Returned by `/read` when called on a `data_root` whose chunks aren't
+    /// `DEFAULT_CHUNK_SIZE`-aligned. The fixed-size offset arithmetic would
+    /// silently serve misaligned bytes; reject instead. Use `GET /content`
+    /// for whole-file reassembly of any root.
+    #[error("data_root is not fixed-size-chunked; use GET /content")]
+    VariableChunkRoot,
 }
 
 #[derive(Serialize)]
@@ -260,6 +267,18 @@ impl IntoResponse for Error {
                     })),
                 },
             ),
+            Error::VariableChunkRoot => (
+                StatusCode::UNPROCESSABLE_ENTITY,
+                ErrorResponse {
+                    error: "variable_chunk_root".to_string(),
+                    details: Some(serde_json::json!({
+                        "message": "this data_root uses variable-size chunks \
+                                    (e.g. CDC); /read assumes DEFAULT_CHUNK_SIZE. \
+                                    Use GET /content?data_root=... for whole-file \
+                                    reassembly."
+                    })),
+                },
+            ),
             Error::NotAcceptingPrimary => (
                 StatusCode::UNPROCESSABLE_ENTITY,
                 ErrorResponse {
diff --git a/provider-node/tests/s3_integration.rs b/provider-node/tests/s3_integration.rs
index c9e90a14..d8b435c6 100644
--- a/provider-node/tests/s3_integration.rs
+++ b/provider-node/tests/s3_integration.rs
@@ -512,6 +512,43 @@ async fn test_get_content_returns_full_bytes() {
     assert_eq!(returned.as_ref(), body.as_slice());
 }
 
+/// `/read` must refuse CDC roots — fixed-size offset math would otherwise
+/// silently serve misaligned bytes.
+#[tokio::test]
+async fn test_read_rejects_cdc_root() {
+    use rand::{RngCore, SeedableRng};
+
+    let server = TestServer::new().await;
+
+    // 2 MiB of random data so CDC emits at least two unequal-sized chunks
+    // (CDC_MIN_SIZE = 64 KiB, CDC_AVG_SIZE = 256 KiB, CDC_MAX_SIZE = 1 MiB).
+    let mut rng = rand::rngs::StdRng::seed_from_u64(0xCDC);
+    let mut body = vec![0u8; 2 * 1024 * 1024];
+    rng.fill_bytes(&mut body);
+
+    let put: Value = server
+        .client
+        .put(server.url("/s3/1/object?key=cdc.bin"))
+        .body(body)
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let data_root = put["data_root"].as_str().unwrap().to_string();
+
+    let r = server
+        .client
+        .get(server.url(&format!("/read?data_root={data_root}&offset=0&length=1024")))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status(), StatusCode::UNPROCESSABLE_ENTITY);
+    let body: Value = r.json().await.unwrap();
+    assert_eq!(body["error"], "variable_chunk_root");
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Edge cases: empty key, nonexistent HEAD/DELETE, empty body, metadata on GET
 // ─────────────────────────────────────────────────────────────────────────────

From 774f2d3c305f10ba861fa3726a843d25f00ea649 Mon Sep 17 00:00:00 2001
From: Anthony Kveder <anthony.kveder@parity.io>
Date: Wed, 1 Jul 2026 08:30:56 +0000
Subject: [PATCH 4/6] Moved h256::from_slice to helper to validate length

---
 docs/design/content-defined-chunking.md |  2 +-
 provider-node/src/api.rs                | 96 +++++++++----------------
 provider-node/tests/api_integration.rs  | 24 +++++++
 3 files changed, 59 insertions(+), 63 deletions(-)

diff --git a/docs/design/content-defined-chunking.md b/docs/design/content-defined-chunking.md
index 5f972bbc..4c771716 100644
--- a/docs/design/content-defined-chunking.md
+++ b/docs/design/content-defined-chunking.md
@@ -38,7 +38,7 @@ The `GET /read?data_root=&offset=&length=` byte-range endpoint assumes fixed-siz
 
 To prevent that, `/read` now **guards** every call: it walks the leaves under `data_root` and rejects with `422 variable_chunk_root` if any non-trailing leaf isn't exactly `DEFAULT_CHUNK_SIZE`. Implementation: `data_root_is_fixed_size` in `provider-node/src/api.rs`; integration coverage in `test_read_rejects_cdc_root`. For whole-file historical fetch, use the new `GET /content?data_root=` endpoint, which walks the manifest and works for any chunking strategy.
 
-Worth flagging: **roots don't record their chunking strategy.** A `data_root` is just a Merkle root over a list of `blake2_256(chunk)` leaves; nothing in the tree distinguishes a fixed-256K-chunked root from a CDC root from anything else. Any future endpoint that does offset/length arithmetic is therefore globally unsafe until we either (a) tag roots with their strategy (e.g. an extra storage map alongside the bucket's S3/FS index), (b) make the read side fully strategy-agnostic by walking a manifest with explicit per-chunk offsets, or (c) only emit one canonical chunking. Right now the guard substitutes for that — by content-inspecting the leaves at request time — and is intentionally conservative (rejects empty roots, rejects anything where any non-trailing leaf differs from `DEFAULT_CHUNK_SIZE`).
+Worth flagging: **roots don't record their chunking strategy.** A `data_root` is just a Merkle root over a list of `blake2_256(chunk)` leaves; nothing in the tree distinguishes a fixed-256K-chunked root from a CDC root from anything else. Any future endpoint that does offset/length arithmetic is therefore globally unsafe until we either (a) tag roots with their strategy (e.g. an extra storage map alongside the bucket's S3/FS index), (b) make the read side fully strategy-agnostic by walking a manifest with explicit per-chunk offsets, or (c) only emit one canonical chunking. Right now the guard substitutes for that by content-inspecting the leaves at request time: it rejects only when it can positively prove the root is variable-size (any non-trailing leaf differs from `DEFAULT_CHUNK_SIZE`). Empty / unknown roots pass through — downstream `get_chunk_at_index` returns empty results cleanly for those, so this preserves the pre-guard behaviour for nonexistent roots.
 
 A real variable-size byte-range implementation of `/read` is deferred until a caller needs it.
 
diff --git a/provider-node/src/api.rs b/provider-node/src/api.rs
index 8c204b8b..205dbe00 100644
--- a/provider-node/src/api.rs
+++ b/provider-node/src/api.rs
@@ -160,6 +160,26 @@ pub(crate) async fn check_role(
     .await
 }
 
+/// Parse a user-supplied hex string into an `H256`. Accepts `0x`-prefixed or
+/// bare hex, validates the character set and enforces the 32-byte length.
+/// Callers previously did `hex_decode(...); H256::from_slice(&bytes)`, which
+/// panics on wrong-length input — this helper returns `Error::InvalidHash`
+/// instead so `?` yields a 400 to the client.
+fn parse_h256(input: &str) -> Result<H256, Error> {
+    let hex = input.strip_prefix("0x").unwrap_or(input);
+    let bytes = hex_decode(hex).map_err(|_| Error::InvalidHash {
+        expected: input.to_string(),
+        actual: "invalid hex".to_string(),
+    })?;
+    if bytes.len() != 32 {
+        return Err(Error::InvalidHash {
+            expected: input.to_string(),
+            actual: format!("wrong length: expected 32 bytes, got {}", bytes.len()),
+        });
+    }
+    Ok(H256::from_slice(&bytes))
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Health and Info
 // ─────────────────────────────────────────────────────────────────────────────
@@ -218,11 +238,7 @@ async fn get_node(
     State(state): State<Arc<ProviderState>>,
     Query(query): Query<GetNodeQuery>,
 ) -> Result<Json<DownloadNodeResponse>, Error> {
-    let hash_bytes = hex_decode(&query.hash).map_err(|_| Error::InvalidHash {
-        expected: query.hash.clone(),
-        actual: "invalid hex".to_string(),
-    })?;
-    let hash = H256::from_slice(&hash_bytes);
+    let hash = parse_h256(&query.hash)?;
 
     let node = state
         .storage
@@ -244,30 +260,17 @@ async fn upload_node(
     State(state): State<Arc<ProviderState>>,
     Json(request): Json<UploadNodeRequest>,
 ) -> Result<Json<UploadNodeResponse>, Error> {
-    // Decode hash
-    let hash_bytes = hex_decode(&request.hash).map_err(|_| Error::InvalidHash {
-        expected: request.hash.clone(),
-        actual: "invalid hex".to_string(),
-    })?;
-    let hash = H256::from_slice(&hash_bytes);
+    let hash = parse_h256(&request.hash)?;
 
-    // Decode data
     let data = BASE64
         .decode(&request.data)
         .map_err(|e| Error::Serialization(e.to_string()))?;
 
-    // Decode children
     let children = request
         .children
         .map(|c| {
             c.iter()
-                .map(|h| {
-                    let bytes = hex_decode(h).map_err(|_| Error::InvalidHash {
-                        expected: h.clone(),
-                        actual: "invalid hex".to_string(),
-                    })?;
-                    Ok(H256::from_slice(&bytes))
-                })
+                .map(|h| parse_h256(h))
                 .collect::<Result<Vec<_>, Error>>()
         })
         .transpose()?;
@@ -290,13 +293,7 @@ async fn check_exists(
     let hashes: Vec<H256> = request
         .hashes
         .iter()
-        .map(|h| {
-            let bytes = hex_decode(h).map_err(|_| Error::InvalidHash {
-                expected: h.clone(),
-                actual: "invalid hex".to_string(),
-            })?;
-            Ok(H256::from_slice(&bytes))
-        })
+        .map(|h| parse_h256(h))
         .collect::<Result<Vec<_>, Error>>()?;
 
     let (exists, missing) = state.storage.check_exists(request.bucket_id, &hashes);
@@ -324,13 +321,7 @@ async fn commit(
     let data_roots: Vec<H256> = request
         .data_roots
         .iter()
-        .map(|h| {
-            let bytes = hex_decode(h).map_err(|_| Error::InvalidHash {
-                expected: h.clone(),
-                actual: "invalid hex".to_string(),
-            })?;
-            Ok(H256::from_slice(&bytes))
-        })
+        .map(|h| parse_h256(h))
         .collect::<Result<Vec<_>, Error>>()?;
 
     let (mmr_root, start_seq, leaf_indices) =
@@ -353,11 +344,7 @@ async fn read_chunks(
     State(state): State<Arc<ProviderState>>,
     Query(query): Query<ReadQuery>,
 ) -> Result<Json<ReadResponse>, Error> {
-    let root_bytes = hex_decode(&query.data_root).map_err(|_| Error::InvalidHash {
-        expected: query.data_root.clone(),
-        actual: "invalid hex".to_string(),
-    })?;
-    let data_root = H256::from_slice(&root_bytes);
+    let data_root = parse_h256(&query.data_root)?;
 
     // `/read`'s offset/length math assumes every leaf except the last is exactly
     // `DEFAULT_CHUNK_SIZE` bytes. CDC-chunked roots break that assumption — the
@@ -400,15 +387,16 @@ async fn read_chunks(
 /// `DEFAULT_CHUNK_SIZE` bytes. Used by `/read` to reject CDC roots — the
 /// fixed-size offset math would otherwise serve misaligned bytes.
 ///
-/// Returns `Ok(false)` for an empty / unknown root (no leaves to validate);
-/// downstream `get_chunk_at_index` already errors cleanly on those.
+/// Unknown / empty roots return `Ok(true)` (pass-through): downstream
+/// `get_chunk_at_index` already returns empty results cleanly for those, and
+/// we don't want to conflate "we can't tell" with "positively CDC."
 fn data_root_is_fixed_size(
     storage: &dyn crate::storage::StorageBackend,
     data_root: H256,
 ) -> Result<bool, Error> {
     let hashes = storage.collect_chunk_hashes(data_root);
     if hashes.is_empty() {
-        return Ok(false);
+        return Ok(true);
     }
     let chunk_size = storage_primitives::DEFAULT_CHUNK_SIZE as usize;
     let (last, prefix) = hashes
@@ -444,11 +432,7 @@ async fn get_content(
     State(state): State<Arc<ProviderState>>,
     Query(query): Query<GetContentQuery>,
 ) -> Result<Response, Error> {
-    let root_bytes = hex_decode(&query.data_root).map_err(|_| Error::InvalidHash {
-        expected: query.data_root.clone(),
-        actual: "invalid hex".to_string(),
-    })?;
-    let data_root = H256::from_slice(&root_bytes);
+    let data_root = parse_h256(&query.data_root)?;
 
     let chunks = state.storage.collect_chunks(data_root);
     if chunks.is_empty() && data_root != H256::zero() {
@@ -563,11 +547,7 @@ async fn get_chunk_proof(
     State(state): State<Arc<ProviderState>>,
     Query(query): Query<ChunkProofQuery>,
 ) -> Result<Json<ChunkProofResponse>, Error> {
-    let root_bytes = hex_decode(&query.data_root).map_err(|_| Error::InvalidHash {
-        expected: query.data_root.clone(),
-        actual: "invalid hex".to_string(),
-    })?;
-    let data_root = H256::from_slice(&root_bytes);
+    let data_root = parse_h256(&query.data_root)?;
 
     let (chunk_data, proof) = state
         .storage
@@ -668,11 +648,7 @@ async fn fetch_nodes(
     let mut nodes = Vec::new();
 
     for hash_str in &request.hashes {
-        let hash_bytes = hex_decode(hash_str).map_err(|_| Error::InvalidHash {
-            expected: hash_str.clone(),
-            actual: "invalid hex".to_string(),
-        })?;
-        let hash = H256::from_slice(&hash_bytes);
+        let hash = parse_h256(hash_str)?;
 
         if let Some(node) = state.storage.get_node(&hash) {
             nodes.push(FetchedNode {
@@ -711,11 +687,7 @@ async fn sign_checkpoint_proposal(
     let local_mmr_root = format!("0x{}", hex_encode(bucket.mmr_root.as_bytes()));
 
     // Check if we agree with the proposal
-    let proposed_root_bytes = hex_decode(&request.mmr_root).map_err(|_| Error::InvalidHash {
-        expected: request.mmr_root.clone(),
-        actual: "invalid hex".to_string(),
-    })?;
-    let proposed_root = H256::from_slice(&proposed_root_bytes);
+    let proposed_root = parse_h256(&request.mmr_root)?;
 
     // We agree if MMR roots match and sequence numbers are compatible
     let agreed = bucket.mmr_root == proposed_root
diff --git a/provider-node/tests/api_integration.rs b/provider-node/tests/api_integration.rs
index d51ad3b5..85c0726a 100644
--- a/provider-node/tests/api_integration.rs
+++ b/provider-node/tests/api_integration.rs
@@ -1393,6 +1393,30 @@ async fn test_chunk_proof_invalid_hex() {
     assert_eq!(body["error"], "invalid_hash");
 }
 
+/// Wrong-length hex on an H256 query parameter used to panic in
+/// `H256::from_slice`, surfacing as a 500. It's now a clean 400 via the
+/// shared `parse_h256` helper.
+#[tokio::test]
+async fn test_short_hex_returns_400_not_500() {
+    let server = TestServer::new().await;
+
+    for path in [
+        "/node?hash=abcd",
+        "/content?data_root=abcd",
+        "/read?data_root=abcd&offset=0&length=10",
+        "/chunk_proof?data_root=abcd&chunk_index=0",
+    ] {
+        let resp = server.client.get(server.url(path)).send().await.unwrap();
+        assert_eq!(
+            resp.status(),
+            StatusCode::BAD_REQUEST,
+            "{path} should be 400"
+        );
+        let body: Value = resp.json().await.unwrap();
+        assert_eq!(body["error"], "invalid_hash", "{path} wrong error tag");
+    }
+}
+
 #[tokio::test]
 async fn test_checkpoint_trigger_unknown_bucket() {
     let server = TestServer::new().await;

From 6ff98ccd1adf22089d223fccc2ea800e86c79b0f Mon Sep 17 00:00:00 2001
From: Anthony Kveder <anthony.kveder@parity.io>
Date: Wed, 1 Jul 2026 08:35:02 +0000
Subject: [PATCH 5/6] Fixed docs

---
 docs/design/content-defined-chunking.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/design/content-defined-chunking.md b/docs/design/content-defined-chunking.md
index 4c771716..a0d86acd 100644
--- a/docs/design/content-defined-chunking.md
+++ b/docs/design/content-defined-chunking.md
@@ -4,9 +4,9 @@
 
 The provider stores file bytes as content-addressed chunks: each chunk's key is `blake2_256(bytes)`, so two PUTs that produce a byte-identical chunk get stored once. That dedup is automatic — *if* the chunker actually produces identical chunks across edits.
 
-With a fixed-size chunker, it doesn't. Inserting a single byte at the start of a file shifts every downstream byte by one, so every chunk has different bytes, so every chunk has a different hash. v2's PUT effectively re-uploads the entire file.
+With a fixed-size chunker, it doesn't. Consider an already-uploaded file (the "original blob") and a small edit that produces a new blob to upload. If the edit inserts a single byte at the start of the file, every downstream byte shifts by one — so every chunk has different bytes, so every chunk has a different hash. The PUT for the new blob effectively re-uploads the entire file.
 
-Content-defined chunking solves this by choosing chunk boundaries from the *content* (via a rolling hash), not from fixed byte offsets. When you insert bytes in the middle of a file, the boundaries downstream of the edit fall at the same content positions as before; the chunks between them are byte-identical to v1's chunks and dedup through content addressing.
+Content-defined chunking solves this by choosing chunk boundaries from the *content* (via a rolling hash), not from fixed byte offsets. When you insert bytes in the middle of a file, the boundaries downstream of the edit fall at the same content positions as before; the chunks between them are byte-identical to the original blob's chunks and dedup through content addressing.
 
 ## Algorithm
 

From 75cc48d4b8e78d500711dbe6f25c77457200c93c Mon Sep 17 00:00:00 2001
From: Anthony Kveder <anthony.kveder@parity.io>
Date: Wed, 1 Jul 2026 10:56:15 +0000
Subject: [PATCH 6/6] Added streaming, made SDK use CDC by default

---
 Cargo.lock                              |   1 +
 Cargo.toml                              |   1 +
 client/src/base.rs                      |  23 ++--
 client/src/storage_user.rs              | 155 +++++++++++++++++-------
 docs/design/content-defined-chunking.md |  10 +-
 provider-node/Cargo.toml                |   1 +
 provider-node/src/api.rs                |  38 ++++--
 provider-node/tests/s3_integration.rs   |  16 +++
 8 files changed, 171 insertions(+), 74 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index c5f7b76b..995adb0b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9794,6 +9794,7 @@ dependencies = [
  "thiserror 2.0.18",
  "tokio",
  "tokio-rate-limit",
+ "tokio-stream",
  "tower-http",
  "tracing",
  "tracing-subscriber 0.3.19",
diff --git a/Cargo.toml b/Cargo.toml
index 02ef3c14..eee0a0f6 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -140,6 +140,7 @@ tracing-subscriber = { version = "=0.3.19" }
 axum = "0.7"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 tokio = { version = "1.0", features = ["full"] }
+tokio-stream = "0.1"
 tower = "0.5"
 tower-http = { version = "0.6", features = ["cors", "trace"] }
 
diff --git a/client/src/base.rs b/client/src/base.rs
index e82265e1..594f6d4a 100644
--- a/client/src/base.rs
+++ b/client/src/base.rs
@@ -211,25 +211,18 @@ pub struct AgreementInfo {
 }
 
 /// Chunking strategy for data upload.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Default)]
 pub enum ChunkingStrategy {
-    /// Fixed-size chunks (default: 256 KiB)
+    /// Fixed-size chunks of the given size. A byte-shifting edit invalidates
+    /// every chunk after the edit point, so cross-version dedup only survives
+    /// in-place overwrites and appends.
     Fixed(usize),
-    /// TODO: Content-defined chunking (not yet implemented)
-    ///
-    /// ContentDefined {
-    ///     min_size: usize,
-    ///     target_size: usize,
-    ///     max_size: usize,
-    /// },
+    /// Content-defined chunks via FastCDC (parameters in
+    /// `storage_primitives::chunking`). Boundaries track content, so an
+    /// insertion only changes the chunks straddling the edit — the default.
+    #[default]
     ContentDefined,
 }
 
-impl Default for ChunkingStrategy {
-    fn default() -> Self {
-        Self::Fixed(256 * 1024)
-    }
-}
-
 /// Result type for client operations.
 pub type ClientResult<T> = Result<T, ClientError>;
diff --git a/client/src/storage_user.rs b/client/src/storage_user.rs
index a006b9ac..c305a373 100644
--- a/client/src/storage_user.rs
+++ b/client/src/storage_user.rs
@@ -14,7 +14,7 @@ use crate::encryption::{Cipher, EncryptionKey, XChaCha20Poly1305Cipher};
 use crate::verification::ClientVerifier;
 use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
 use sp_core::H256;
-use storage_primitives::{blake2_256, BucketId};
+use storage_primitives::{blake2_256, verify_merkle_proof, BucketId, MerkleProof};
 
 /// Client for storage users (end users who store/retrieve data).
 pub struct StorageUserClient {
@@ -169,20 +169,72 @@ impl StorageUserClient {
         offset: u64,
         length: u64,
     ) -> ClientResult<Vec<u8>> {
+        // Walk chunks by index, verifying each against `data_root`, until we've
+        // covered the requested range or the provider signals end-of-file.
+        // Content-defined chunking makes chunk sizes variable, so a byte offset
+        // can't be mapped to a chunk index up front — we walk from chunk 0. In
+        // practice callers read from offset 0 (whole object or prefix), so this
+        // is a single forward pass; `spot_check` fetches a single chunk by index
+        // directly rather than going through here.
+        let end = offset.saturating_add(length);
+        let mut data = Vec::new();
+        let mut chunk_index = 0u64;
+        while (data.len() as u64) < end {
+            match self.fetch_chunk_verified(data_root, chunk_index).await? {
+                Some(chunk) => data.extend_from_slice(&chunk),
+                None => break, // no more chunks
+            }
+            chunk_index += 1;
+        }
+
+        // Slice to the requested range (clamped — a short object yields fewer
+        // bytes rather than erroring).
+        let start = (offset as usize).min(data.len());
+        let stop = (end as usize).min(data.len());
+        let ranged = data[start..stop].to_vec();
+
+        // Decrypt after reassembly if encryption is enabled.
+        if let Some(cipher) = &self.cipher {
+            cipher.decrypt(&ranged)
+        } else {
+            Ok(ranged)
+        }
+    }
+
+    /// Fetch a single chunk by index and verify its Merkle proof against
+    /// `data_root`. Returns `Ok(None)` when `chunk_index` is past the last
+    /// chunk (the provider's 404 end-of-file signal), so callers can iterate
+    /// `0..` until they hit it.
+    ///
+    /// Uses `/chunk_proof`, which is index-based and therefore works for any
+    /// chunking strategy — unlike `/read`, whose fixed-size offset math the
+    /// provider rejects for content-defined roots. This is also strictly
+    /// stronger than the old `/read` path: it verifies the chunk actually sits
+    /// under `data_root`, not just that the bytes match a hash the provider
+    /// itself supplied.
+    async fn fetch_chunk_verified(
+        &self,
+        data_root: &H256,
+        chunk_index: u64,
+    ) -> ClientResult<Option<Vec<u8>>> {
         let provider_url = self.base.get_provider_url()?;
 
         let response = self
             .base
             .http
-            .get(format!("{provider_url}/read"))
+            .get(format!("{provider_url}/chunk_proof"))
             .query(&[
                 ("data_root", BaseClient::hex_encode(data_root.as_bytes())),
-                ("offset", offset.to_string()),
-                ("length", length.to_string()),
+                ("chunk_index", chunk_index.to_string()),
             ])
             .send()
             .await?;
 
+        // 404 means `chunk_index` ran past the last leaf — the loop's stop
+        // condition, not an error.
+        if response.status() == reqwest::StatusCode::NOT_FOUND {
+            return Ok(None);
+        }
         if !response.status().is_success() {
             return Err(ClientError::Api(format!(
                 "Provider returned error: {}",
@@ -190,40 +242,51 @@ impl StorageUserClient {
             )));
         }
 
-        let read_response: ReadResponse = response.json().await?;
-
-        let mut data = Vec::new();
-        for chunk in read_response.chunks {
-            let chunk_data = BASE64
-                .decode(&chunk.data)
-                .map_err(|e| ClientError::Serialization(e.to_string()))?;
-
-            // Verify chunk hash
-            let expected_hash = BaseClient::hex_decode(&chunk.hash)?;
-            let actual_hash = blake2_256(&chunk_data);
-            if actual_hash.as_bytes() != expected_hash.as_slice() {
-                return Err(ClientError::VerificationFailed);
-            }
+        let proof_response: ChunkProofResponse = response
+            .json()
+            .await
+            .map_err(|e| ClientError::Serialization(e.to_string()))?;
 
-            data.extend_from_slice(&chunk_data);
+        let chunk_data = proof_response
+            .chunk_data
+            .ok_or(ClientError::VerificationFailed)
+            .and_then(|d| {
+                BASE64
+                    .decode(d)
+                    .map_err(|e| ClientError::Serialization(e.to_string()))
+            })?;
+
+        // 1. The leaf hash must be the hash of the bytes we actually received.
+        let leaf_hash = blake2_256(&chunk_data);
+        let claimed_hash = BaseClient::hex_decode(&proof_response.chunk_hash)?;
+        if leaf_hash.as_bytes() != claimed_hash.as_slice() {
+            return Err(ClientError::VerificationFailed);
         }
 
-        // Trim to requested range
-        let chunk_size = 256 * 1024;
-        let start = (offset % chunk_size) as usize;
-        let end = start + length as usize;
-        let trimmed = if end <= data.len() {
-            data[start..end].to_vec()
-        } else {
-            data[start..].to_vec()
+        // 2. That leaf must sit under `data_root` at `chunk_index`.
+        let siblings = proof_response
+            .proof
+            .siblings
+            .iter()
+            .map(|h| {
+                let bytes = BaseClient::hex_decode(h)?;
+                if bytes.len() != 32 {
+                    return Err(ClientError::Serialization(
+                        "proof sibling is not 32 bytes".to_string(),
+                    ));
+                }
+                Ok(H256::from_slice(&bytes))
+            })
+            .collect::<ClientResult<Vec<H256>>>()?;
+        let proof = MerkleProof {
+            siblings,
+            path: proof_response.proof.path,
         };
-
-        // Decrypt after reassembly if encryption is enabled
-        if let Some(cipher) = &self.cipher {
-            cipher.decrypt(&trimmed)
-        } else {
-            Ok(trimmed)
+        if !verify_merkle_proof(leaf_hash, chunk_index, &proof, data_root) {
+            return Err(ClientError::VerificationFailed);
         }
+
+        Ok(Some(chunk_data))
     }
 
     /// Download entire file by data root.
@@ -403,22 +466,22 @@ impl StorageUserClient {
     pub async fn spot_check(&mut self, data_root: &H256, chunk_index: u64) -> ClientResult<bool> {
         use std::time::Instant;
 
-        let chunk_size = 256 * 1024u64; // 256 KiB
-        let offset = chunk_index * chunk_size;
         let provider_url = self.base.get_provider_url()?.to_string();
 
+        // Fetch the chunk by index and verify its Merkle proof against
+        // `data_root`. Index-based, so it's correct regardless of chunk size.
         let start = Instant::now();
-        let result = self.download(data_root, offset, chunk_size).await;
+        let result = self.fetch_chunk_verified(data_root, chunk_index).await;
         let duration = start.elapsed();
 
         match result {
-            Ok(data) if !data.is_empty() => {
+            Ok(Some(_)) => {
                 self.verifier.record_request(&provider_url, duration, true);
                 Ok(true)
             }
             _ => {
-                // Either download failed or the provider returned no data for this chunk,
-                // which means the data is unavailable.
+                // Missing chunk, failed verification, or transport error — the
+                // provider can't prove it holds this chunk.
                 self.verifier.record_request(&provider_url, duration, false);
                 Ok(false)
             }
@@ -683,16 +746,16 @@ pub struct CommitResponse {
 }
 
 #[derive(serde::Deserialize)]
-struct ReadResponse {
-    chunks: Vec<ChunkWithProof>,
+struct ChunkProofResponse {
+    chunk_hash: String,
+    chunk_data: Option<String>,
+    proof: MerkleProofResponse,
 }
 
 #[derive(serde::Deserialize)]
-#[allow(dead_code)]
-struct ChunkWithProof {
-    hash: String,
-    data: String,
-    proof: Vec<String>,
+struct MerkleProofResponse {
+    siblings: Vec<String>,
+    path: Vec<bool>,
 }
 
 #[derive(serde::Deserialize)]
diff --git a/docs/design/content-defined-chunking.md b/docs/design/content-defined-chunking.md
index a0d86acd..203f3562 100644
--- a/docs/design/content-defined-chunking.md
+++ b/docs/design/content-defined-chunking.md
@@ -36,7 +36,7 @@ These live in [`primitives/src/chunking.rs`](../../primitives/src/chunking.rs).
 
 The `GET /read?data_root=&offset=&length=` byte-range endpoint assumes fixed-size chunks — its arithmetic is `chunk_index = offset / DEFAULT_CHUNK_SIZE`. Now that S3 and FS uploads always use CDC, that's wrong for the **common** case rather than a rare one, and `get_chunk_at_index` would still hand back *a* chunk, so the endpoint would silently serve misaligned bytes with a 200.
 
-To prevent that, `/read` now **guards** every call: it walks the leaves under `data_root` and rejects with `422 variable_chunk_root` if any non-trailing leaf isn't exactly `DEFAULT_CHUNK_SIZE`. Implementation: `data_root_is_fixed_size` in `provider-node/src/api.rs`; integration coverage in `test_read_rejects_cdc_root`. For whole-file historical fetch, use the new `GET /content?data_root=` endpoint, which walks the manifest and works for any chunking strategy.
+To prevent that, `/read` now **guards** every call: it walks the leaves under `data_root` and rejects with `422 variable_chunk_root` if any non-trailing leaf isn't exactly `DEFAULT_CHUNK_SIZE`. Implementation: `data_root_is_fixed_size` in `provider-node/src/api.rs`; integration coverage in `test_read_rejects_cdc_root`. For whole-file historical fetch, use the `GET /content?data_root=` endpoint, which walks the tree by hash (streaming chunk-by-chunk, so provider memory is one chunk per request, not the whole object) and works for any chunking strategy.
 
 Worth flagging: **roots don't record their chunking strategy.** A `data_root` is just a Merkle root over a list of `blake2_256(chunk)` leaves; nothing in the tree distinguishes a fixed-256K-chunked root from a CDC root from anything else. Any future endpoint that does offset/length arithmetic is therefore globally unsafe until we either (a) tag roots with their strategy (e.g. an extra storage map alongside the bucket's S3/FS index), (b) make the read side fully strategy-agnostic by walking a manifest with explicit per-chunk offsets, or (c) only emit one canonical chunking. Right now the guard substitutes for that by content-inspecting the leaves at request time: it rejects only when it can positively prove the root is variable-size (any non-trailing leaf differs from `DEFAULT_CHUNK_SIZE`). Empty / unknown roots pass through — downstream `get_chunk_at_index` returns empty results cleanly for those, so this preserves the pre-guard behaviour for nonexistent roots.
 
@@ -54,10 +54,14 @@ A real variable-size byte-range implementation of `/read` is deferred until a ca
 | Append dedup | ✅ Works (only trailing chunk changes) | ✅ Works |
 | Best for | Binary blobs with fixed-offset fields; embedded use where every byte counts | Text, structured data, anything with shifting edits |
 
-`ChunkingStrategy::Fixed(usize)` remains the SDK default; CDC is opt-in via `ChunkingStrategy::ContentDefined`. The provider's HTTP upload endpoints (S3 / FS) always use CDC since their callers don't pick a strategy.
+`ChunkingStrategy::default()` is now `ContentDefined` — the SDK CDC-chunks on upload by default; `Fixed(usize)` stays available for the binary/embedded cases above. The provider's HTTP upload endpoints (S3 / FS) always use CDC since their callers don't pick a strategy.
+
+The default could flip because the SDK's read path no longer depends on `/read`. `StorageUserClient::download` (and `spot_check`) now fetch chunks by index through `GET /chunk_proof?data_root=&chunk_index=`, which is index-based — it never touches the fixed-size offset math that the `/read` guard rejects — and returns each chunk with its Merkle proof. The client verifies every chunk two ways: `blake2_256(bytes) == leaf_hash`, then `verify_merkle_proof(leaf_hash, index, proof, data_root)`. It iterates `0..` until the provider 404s (end of file). This is strictly stronger than the old `/read` path, which returned proofs but never checked them against `data_root` — so a lying provider was previously undetectable on download. `download(offset, length)` walks from chunk 0 and slices the result; there's no client-side random access by byte offset under CDC (chunk sizes are content-defined), which is fine since real callers read from offset 0.
 
 ## Verification
 
 - Unit tests in `primitives/src/chunking.rs::tests` cover determinism, size-distribution bounds, byte-equal reassembly, and ≥ 90% chunk reuse on mid-file insertion / deletion of 8 MiB random data.
 - Integration test `test_s3_cdc_dedup_across_versions` in `provider-node/tests/s3_integration.rs` PUTs an 8 MiB blob, then a mid-file-edited variant, and asserts the second PUT adds fewer than `total_nodes(v1) / 4` new nodes.
-- `test_get_content_returns_full_bytes` exercises the new `/content` endpoint.
+- `test_get_content_returns_full_bytes` exercises `/content` (multi-chunk); `test_get_content_unknown_root_404` covers the missing-root path.
+- `test_read_rejects_cdc_root` asserts `/read` refuses variable-size roots.
+- The SDK's `download` / `spot_check` verify each chunk's Merkle proof against `data_root` via `/chunk_proof`, so a CDC `upload` → `download` round-trip is cryptographically checked end-to-end.
diff --git a/provider-node/Cargo.toml b/provider-node/Cargo.toml
index e15c67eb..8f1704db 100644
--- a/provider-node/Cargo.toml
+++ b/provider-node/Cargo.toml
@@ -11,6 +11,7 @@ description = "Off-chain provider node for scalable Web3 storage"
 storage-client = { workspace = true }
 storage-primitives = { workspace = true, features = ["serde", "std"] }
 tokio = { workspace = true }
+tokio-stream = { workspace = true }
 axum = { workspace = true }
 tower-http = { workspace = true }
 tokio-rate-limit = { version = "0.8" }
diff --git a/provider-node/src/api.rs b/provider-node/src/api.rs
index 205dbe00..1b00bee9 100644
--- a/provider-node/src/api.rs
+++ b/provider-node/src/api.rs
@@ -14,10 +14,11 @@ use crate::storage::{hex_decode, hex_encode};
 use crate::types::*;
 use crate::ProviderState;
 use axum::{
+    body::{Body, Bytes},
     extract::{ConnectInfo, DefaultBodyLimit, Query, Request, State},
-    http::{header, HeaderValue, StatusCode},
+    http::{header, HeaderValue},
     middleware::{from_fn_with_state, Next},
-    response::{IntoResponse, Response},
+    response::Response,
     routing::{get, post, put},
     Json, Router,
 };
@@ -434,16 +435,33 @@ async fn get_content(
 ) -> Result<Response, Error> {
     let data_root = parse_h256(&query.data_root)?;
 
-    let chunks = state.storage.collect_chunks(data_root);
-    if chunks.is_empty() && data_root != H256::zero() {
+    // Resolve the leaf list up front (cheap — hashes only) so a missing root
+    // returns a clean 404 before we commit to a 200 + streamed body.
+    let chunk_hashes = state.storage.collect_chunk_hashes(data_root);
+    if chunk_hashes.is_empty() && data_root != H256::zero() {
         return Err(Error::NodeNotFound(query.data_root));
     }
-    let total: usize = chunks.iter().map(Vec::len).sum();
-    let mut body = Vec::with_capacity(total);
-    for chunk in chunks {
-        body.extend_from_slice(&chunk);
-    }
-    let mut response = (StatusCode::OK, body).into_response();
+
+    // Stream chunk-by-chunk instead of buffering the whole object: peak memory
+    // is one chunk per in-flight request, not object-size × concurrency. Each
+    // leaf's bytes are fetched from storage on demand as the body is polled.
+    let storage = state.storage.clone();
+    let stream = tokio_stream::iter(chunk_hashes.into_iter().map(move |hash| {
+        storage
+            .get_node(&hash)
+            .map(|node| Bytes::from(node.data))
+            .ok_or_else(|| {
+                // A leaf vanished mid-stream (e.g. concurrent prune). The 200 is
+                // already sent, so truncate the body with an error rather than
+                // silently serving a short read.
+                std::io::Error::other(format!(
+                    "chunk 0x{} missing during stream",
+                    hex_encode(hash.as_bytes())
+                ))
+            })
+    }));
+
+    let mut response = Response::new(Body::from_stream(stream));
     response.headers_mut().insert(
         header::CONTENT_TYPE,
         HeaderValue::from_static("application/octet-stream"),
diff --git a/provider-node/tests/s3_integration.rs b/provider-node/tests/s3_integration.rs
index d8b435c6..d2974816 100644
--- a/provider-node/tests/s3_integration.rs
+++ b/provider-node/tests/s3_integration.rs
@@ -512,6 +512,22 @@ async fn test_get_content_returns_full_bytes() {
     assert_eq!(returned.as_ref(), body.as_slice());
 }
 
+/// A well-formed but unknown `data_root` must still 404 — the emptiness check
+/// now runs against `collect_chunk_hashes` before the streaming body starts.
+#[tokio::test]
+async fn test_get_content_unknown_root_404() {
+    let server = TestServer::new().await;
+
+    let unknown = format!("0x{}", "ab".repeat(32));
+    let r = server
+        .client
+        .get(server.url(&format!("/content?data_root={unknown}")))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status(), StatusCode::NOT_FOUND);
+}
+
 /// `/read` must refuse CDC roots — fixed-size offset math would otherwise
 /// silently serve misaligned bytes.
 #[tokio::test]