claw-store/src/cluster/blob.rs

@@ -353,6 +353,90 @@ impl BlobStore {
         Ok(true)
     }
 
+    // ── Chunk-level API (Phase 2d) ───────────────────────────────────
+    //
+    // These operations expose the store's chunk substrate directly so a
+    // sender can query + upload one chunk at a time. Combined with
+    // gossip-derived affinity + the `HasChunk` RPC, they let a peer
+    // replicate a blob by transferring only chunks the receiver is
+    // missing — the "big win when caches overlap" case.
+
+    /// Whether a chunk file exists on disk. No hash verification —
+    /// callers who want cryptographic proof should follow with
+    /// [`read_chunk`] and check the returned hash themselves.
+    pub async fn has_chunk(&self, hash: &ChunkHash) -> Result<bool> {
+        match tokio::fs::metadata(self.chunk_path(hash)).await {
+            Ok(_) => Ok(true),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+            Err(e) => Err(anyhow::Error::from(e)),
+        }
+    }
+
+    /// Read a single chunk's raw bytes. Returns `None` when the chunk
+    /// isn't on disk. Verifies the chunk hashes to the requested
+    /// `hash` (defense in depth against silent corruption); a mismatch
+    /// returns `Err`.
+    pub async fn read_chunk(&self, hash: &ChunkHash) -> Result<Option<Vec<u8>>> {
+        let path = self.chunk_path(hash);
+        let bytes = match tokio::fs::read(&path).await {
+            Ok(b) => b,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(e) => return Err(anyhow::Error::from(e)),
+        };
+        let recomputed = ChunkHash(blake3::hash(&bytes).into());
+        if recomputed != *hash {
+            bail!(
+                "chunk hash mismatch at {}: requested {}, disk hashes to {}",
+                path.display(),
+                hash.to_hex(),
+                recomputed.to_hex()
+            );
+        }
+        Ok(Some(bytes))
+    }
+
+    /// Store one chunk. Verifies that `bytes` actually hash to `hash`
+    /// before writing — the caller can't accidentally (or maliciously)
+    /// stash unrelated bytes under a hash it doesn't own. Idempotent:
+    /// writing the same chunk twice is a no-op after the first.
+    pub async fn put_chunk(&self, hash: &ChunkHash, bytes: &[u8]) -> Result<()> {
+        let recomputed = ChunkHash(blake3::hash(bytes).into());
+        if recomputed != *hash {
+            bail!(
+                "chunk hash mismatch: claimed {}, bytes hash to {}",
+                hash.to_hex(),
+                recomputed.to_hex()
+            );
+        }
+        self.write_chunk_if_absent(hash, bytes).await
+    }
+
+    /// Persist a manifest whose referenced chunks are already on disk.
+    /// Verifies that every chunk in the manifest exists before writing —
+    /// callers can't accidentally register a manifest that references
+    /// missing chunks (which would break subsequent `get_bytes`).
+    ///
+    /// Returns the list of chunk hashes that were missing when the
+    /// call started; if empty, the manifest was written. Otherwise
+    /// nothing was written and the client is expected to upload the
+    /// missing chunks (typically via [`put_chunk`]) and retry.
+    pub async fn put_manifest_verified(
+        &self,
+        manifest: &BlobManifest,
+    ) -> Result<Vec<ChunkHash>> {
+        let mut missing = Vec::new();
+        for chunk_hash in &manifest.chunks {
+            if !self.has_chunk(chunk_hash).await? {
+                missing.push(*chunk_hash);
+            }
+        }
+        if !missing.is_empty() {
+            return Ok(missing);
+        }
+        self.write_manifest_if_absent(manifest).await?;
+        Ok(Vec::new())
+    }
+
     /// Whether a manifest exists for `id`. Cheap — no chunk reads.
     pub async fn contains(&self, id: &BlobId) -> Result<bool> {
         match tokio::fs::metadata(self.manifest_path(id)).await {
@@ -471,7 +555,10 @@ impl BlobStore {
             .join(format!("{hex}.manifest.json"))
     }
 
-    fn chunk_path(&self, hash: &ChunkHash) -> PathBuf {
+    /// On-disk path for a chunk. Public so tests + advanced callers
+    /// (e.g. custom migration tools) can address individual chunks
+    /// without duplicating the layout convention.
+    pub fn chunk_path(&self, hash: &ChunkHash) -> PathBuf {
         let hex = hash.to_hex();
         self.root.join("chunks").join(&hex[..2]).join(&hex)
     }
@@ -959,6 +1046,103 @@ mod tests {
         assert!(sink.is_empty());
     }
 
+    // ── Phase 2d: chunk-level API ────────────────────────────────────
+
+    #[tokio::test]
+    async fn has_chunk_is_false_before_put_and_true_after() {
+        let (_tmp, store) = open_store();
+        let bytes = b"chunk-content";
+        let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
+        assert!(!store.has_chunk(&hash).await.unwrap());
+        store.put_chunk(&hash, bytes).await.unwrap();
+        assert!(store.has_chunk(&hash).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn read_chunk_returns_bytes_and_none_when_missing() {
+        let (_tmp, store) = open_store();
+        let bytes = b"read-me";
+        let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
+        assert!(store.read_chunk(&hash).await.unwrap().is_none());
+        store.put_chunk(&hash, bytes).await.unwrap();
+        let round = store.read_chunk(&hash).await.unwrap().unwrap();
+        assert_eq!(round, bytes);
+    }
+
+    #[tokio::test]
+    async fn put_chunk_rejects_hash_mismatch() {
+        let (_tmp, store) = open_store();
+        // Claim a hash that doesn't match the bytes.
+        let bogus = ChunkHash::from_bytes([0xffu8; 32]);
+        let err = store
+            .put_chunk(&bogus, b"real content")
+            .await
+            .unwrap_err()
+            .to_string();
+        assert!(
+            err.contains("chunk hash mismatch"),
+            "unexpected error: {err}"
+        );
+        // Nothing should have been written.
+        assert!(!store.has_chunk(&bogus).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn read_chunk_detects_corruption() {
+        let (_tmp, store) = open_store();
+        let bytes = b"tamperproof";
+        let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
+        store.put_chunk(&hash, bytes).await.unwrap();
+        // Corrupt the on-disk file.
+        std::fs::write(store.chunk_path(&hash), b"tampered").unwrap();
+        let err = store.read_chunk(&hash).await.unwrap_err().to_string();
+        assert!(err.contains("chunk hash mismatch"));
+    }
+
+    #[tokio::test]
+    async fn put_manifest_verified_reports_missing_chunks() {
+        let (_tmp, store) = open_store();
+        // Cook a manifest whose chunks aren't on disk.
+        let phantom_chunks = vec![
+            ChunkHash::from_bytes([0x01u8; 32]),
+            ChunkHash::from_bytes([0x02u8; 32]),
+        ];
+        let manifest = BlobManifest {
+            blob_id: BlobId::from_bytes([0x03u8; 32]),
+            total_size: 100,
+            chunks: phantom_chunks.clone(),
+        };
+        let missing = store.put_manifest_verified(&manifest).await.unwrap();
+        assert_eq!(missing, phantom_chunks);
+        // Manifest must NOT have been persisted since chunks are missing.
+        assert!(!store.contains(&manifest.blob_id).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn put_manifest_verified_writes_when_all_chunks_present() {
+        let (_tmp, store) = open_store();
+        // Upload two chunks, then commit a manifest referencing them.
+        let a = b"chunk-a";
+        let b = b"chunk-b";
+        let hash_a = ChunkHash::from_bytes(blake3::hash(a).into());
+        let hash_b = ChunkHash::from_bytes(blake3::hash(b).into());
+        store.put_chunk(&hash_a, a).await.unwrap();
+        store.put_chunk(&hash_b, b).await.unwrap();
+
+        // BlobId here is arbitrary — chunk-level API doesn't verify
+        // that blob_id = blake3(concat(chunks)); that's the sender's
+        // responsibility (production callers get it right via the
+        // top-level put_bytes / put_stream paths).
+        let manifest = BlobManifest {
+            blob_id: BlobId::from_bytes([0x99u8; 32]),
+            total_size: (a.len() + b.len()) as u64,
+            chunks: vec![hash_a, hash_b],
+        };
+        let missing = store.put_manifest_verified(&manifest).await.unwrap();
+        assert!(missing.is_empty());
+        assert!(store.contains(&manifest.blob_id).await.unwrap());
+    }
+
     #[tokio::test]
     async fn stream_to_detects_chunk_corruption() {
         let (_tmp, store) = open_store();

claw-store/src/cluster/rpc.rs

@@ -25,7 +25,7 @@
 //!   containing this node's local view of the cluster (its own
 //!   name+zone, plus every peer it currently knows about via gossip).
 
-use crate::cluster::blob::{BlobId, BlobManifest, BlobStat, BlobStore};
+use crate::cluster::blob::{BlobId, BlobManifest, BlobStat, BlobStore, ChunkHash};
 use crate::cluster::gossip::{ClusterGossip, PeerView};
 use anyhow::{bail, Context, Result};
 use quinn::{Connection, ConnectionError};
@@ -76,6 +76,27 @@ pub enum Method {
     /// streamed until the server half-closes, OR a single-byte error
     /// code (typically [`ErrorCode::NotFound`]).
     BlobGetStream = 0x08,
+    /// Phase 2d: does the peer already have a specific chunk?
+    /// `payload`: 32-byte `ChunkHash`. Reply: single byte —
+    /// `STREAM_STATUS_OK` (present) or [`ErrorCode::NotFound`] (absent).
+    /// Non-error single-byte replies make it cheap enough to fan out
+    /// N of these during a partial-sync scan.
+    HasChunk = 0x09,
+    /// Phase 2d: upload one chunk. `payload`: 32-byte ChunkHash ||
+    /// chunk bytes. Reply: `STREAM_STATUS_OK` (1 byte) on success, or
+    /// single-byte error code. The server verifies the bytes hash to
+    /// the claimed hash before writing (defense against poisoning).
+    PutChunk = 0x0a,
+    /// Phase 2d: fetch one chunk. `payload`: 32-byte ChunkHash.
+    /// Reply: `STREAM_STATUS_OK` (1 byte) || chunk bytes on success,
+    /// or single-byte [`ErrorCode::NotFound`].
+    GetChunk = 0x0b,
+    /// Phase 2d: commit a blob manifest whose chunks are already on
+    /// the peer's disk. `payload`: JSON `BlobManifest`. Reply: JSON —
+    /// `{"blob_id":"...","missing":[<chunk_hash>...]}` where an empty
+    /// `missing` list means the manifest was written; a non-empty
+    /// list tells the client which chunks to upload before retrying.
+    PutManifest = 0x0c,
 }
 
 impl Method {
@@ -91,6 +112,10 @@ impl Method {
             0x06 => Some(Method::BlobLoadManifest),
             0x07 => Some(Method::BlobPutStream),
             0x08 => Some(Method::BlobGetStream),
+            0x09 => Some(Method::HasChunk),
+            0x0a => Some(Method::PutChunk),
+            0x0b => Some(Method::GetChunk),
+            0x0c => Some(Method::PutManifest),
             _ => None,
         }
     }
@@ -157,6 +182,16 @@ pub struct PeerStatusReply {
     pub peers: Vec<PeerView>,
 }
 
+/// Reply payload for [`Method::PutManifest`]. When `missing` is empty
+/// the manifest was persisted successfully; otherwise the client must
+/// upload the listed chunks (typically via [`Method::PutChunk`]) and
+/// retry the same manifest.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct PutManifestReply {
+    pub blob_id: crate::cluster::blob::BlobId,
+    pub missing: Vec<crate::cluster::blob::ChunkHash>,
+}
+
 /// The concrete RPC handler used by the daemon. Holds Arc references
 /// to the state a request might need to read: the gossip service
 /// (always), and optionally a local blob store (for Blob* methods).
@@ -316,8 +351,95 @@ impl RpcRouter {
                 // error so it's obvious what happened.
                 Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
             }
+            Method::HasChunk => {
+                let store = match &self.blob_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let hash = match decode_chunk_hash(payload) {
+                    Some(h) => h,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                if store.has_chunk(&hash).await? {
+                    Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
+                } else {
+                    Ok(HandlerOutcome::Error(ErrorCode::NotFound))
                 }
             }
+            Method::PutChunk => {
+                let store = match &self.blob_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                if payload.len() < 32 {
+                    return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest));
+                }
+                let mut hash_bytes = [0u8; 32];
+                hash_bytes.copy_from_slice(&payload[..32]);
+                let hash = ChunkHash::from_bytes(hash_bytes);
+                let chunk_bytes = &payload[32..];
+                match store.put_chunk(&hash, chunk_bytes).await {
+                    Ok(()) => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
+                    Err(e) => {
+                        tracing::warn!(error = %e, "PutChunk rejected");
+                        Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
+                    }
+                }
+            }
+            Method::GetChunk => {
+                let store = match &self.blob_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let hash = match decode_chunk_hash(payload) {
+                    Some(h) => h,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                match store.read_chunk(&hash).await? {
+                    Some(bytes) => {
+                        // STREAM_STATUS_OK prefix so a legitimate first
+                        // content byte of 0xf3 isn't confused with
+                        // NotFound. Fixed 1-byte overhead.
+                        let mut reply = Vec::with_capacity(1 + bytes.len());
+                        reply.push(STREAM_STATUS_OK);
+                        reply.extend_from_slice(&bytes);
+                        Ok(HandlerOutcome::Reply(reply))
+                    }
+                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
+                }
+            }
+            Method::PutManifest => {
+                let store = match &self.blob_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let manifest: BlobManifest = match serde_json::from_slice(payload) {
+                    Ok(m) => m,
+                    Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                let missing = store.put_manifest_verified(&manifest).await?;
+                let reply = PutManifestReply {
+                    blob_id: manifest.blob_id,
+                    missing,
+                };
+                let json = serde_json::to_vec(&reply)
+                    .context("encoding PutManifestReply as JSON")?;
+                Ok(HandlerOutcome::Reply(json))
+            }
+        }
+    }
+}
+
+/// Parse a payload as a 32-byte ChunkHash. Same shape as
+/// [`decode_blob_id`] but a distinct function so the reader-visible
+/// type at each call site is unambiguous.
+fn decode_chunk_hash(payload: &[u8]) -> Option<ChunkHash> {
+    if payload.len() != 32 {
+        return None;
+    }
+    let mut buf = [0u8; 32];
+    buf.copy_from_slice(payload);
+    Some(ChunkHash::from_bytes(buf))
 }
 
 /// Parse a payload as a 32-byte BlobId. Returns `None` for any other
@@ -743,6 +865,189 @@ pub async fn call_blob_load_manifest(
     Ok(Some(manifest))
 }
 
+// ── Phase 2d: chunk-level client helpers ─────────────────────────────
+
+/// Does the peer already have this chunk? `Ok(true)` on presence,
+/// `Ok(false)` on NotFound. Other errors surface as `Err`.
+pub async fn call_has_chunk(
+    conn: &Connection,
+    hash: &ChunkHash,
+) -> Result<bool> {
+    let reply = rpc_call(conn, Method::HasChunk, hash.as_bytes()).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte HasChunk reply, got {} bytes",
+            reply.len()
+        );
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(true),
+        code => match decode_error(code) {
+            Some(ErrorCode::NotFound) => Ok(false),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for HasChunk",
+                code
+            ),
+        },
+    }
+}
+
+/// Upload a single chunk to the peer's store. Peer verifies bytes
+/// hash to `hash` before writing; a mismatch surfaces as `Err`.
+pub async fn call_put_chunk(
+    conn: &Connection,
+    hash: &ChunkHash,
+    bytes: &[u8],
+) -> Result<()> {
+    let mut payload = Vec::with_capacity(32 + bytes.len());
+    payload.extend_from_slice(hash.as_bytes());
+    payload.extend_from_slice(bytes);
+    let reply = rpc_call(conn, Method::PutChunk, &payload).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte PutChunk reply, got {} bytes",
+            reply.len()
+        );
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(()),
+        code => match decode_error(code) {
+            Some(err) => bail!("peer rejected chunk: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for PutChunk",
+                code
+            ),
+        },
+    }
+}
+
+/// Fetch a single chunk from the peer's store. `Ok(None)` on NotFound.
+/// Verifies the returned bytes hash to `hash` locally — corruption or
+/// protocol drift surfaces as `Err`.
+pub async fn call_get_chunk(
+    conn: &Connection,
+    hash: &ChunkHash,
+) -> Result<Option<Vec<u8>>> {
+    let reply = rpc_call(conn, Method::GetChunk, hash.as_bytes()).await?;
+    if reply.is_empty() {
+        bail!("empty GetChunk reply");
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => {
+            let bytes = reply[1..].to_vec();
+            let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
+            if recomputed != *hash {
+                bail!(
+                    "GetChunk hash mismatch: requested {}, got bytes hashing to {}",
+                    hash.to_hex(),
+                    recomputed.to_hex()
+                );
+            }
+            Ok(Some(bytes))
+        }
+        code => match decode_error(code) {
+            Some(ErrorCode::NotFound) => Ok(None),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => {
+                if reply.len() == 1 {
+                    bail!(
+                        "peer replied with unknown byte 0x{:02x} for GetChunk",
+                        code
+                    );
+                }
+                // A single legitimate content byte with value 0xf3 is
+                // technically distinguishable from NotFound because
+                // GetChunk always prefixes with STREAM_STATUS_OK. The
+                // unreachable branch stays as belt-and-braces.
+                let bytes = reply[1..].to_vec();
+                let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
+                if recomputed != *hash {
+                    bail!("GetChunk hash mismatch (fallback path)");
+                }
+                Ok(Some(bytes))
+            }
+        },
+    }
+}
+
+/// Commit a manifest whose chunks the peer should already have.
+/// Returns the list of chunk hashes the peer is still missing — an
+/// empty list means the manifest was persisted; a non-empty list
+/// tells the caller which chunks to upload before retrying.
+pub async fn call_put_manifest(
+    conn: &Connection,
+    manifest: &BlobManifest,
+) -> Result<Vec<ChunkHash>> {
+    let payload =
+        serde_json::to_vec(manifest).context("encoding BlobManifest as JSON")?;
+    let reply = rpc_call(conn, Method::PutManifest, &payload).await?;
+    if reply.len() == 1 {
+        if let Some(err) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", err.describe());
+        }
+    }
+    let decoded: PutManifestReply = serde_json::from_slice(&reply)
+        .context("decoding PutManifestReply JSON")?;
+    if decoded.blob_id != manifest.blob_id {
+        bail!(
+            "peer echoed blob_id {} but we sent {}",
+            decoded.blob_id.to_hex(),
+            manifest.blob_id.to_hex()
+        );
+    }
+    Ok(decoded.missing)
+}
+
+/// High-level partial-sync helper: replicate a local blob to a peer
+/// by uploading only chunks the peer is missing.
+///
+/// Flow:
+///   1. Load the local manifest.
+///   2. For each chunk, ask the peer if it already has it.
+///   3. Upload the missing chunks.
+///   4. Send the manifest so the peer commits.
+///
+/// Returns `(uploaded_chunks, total_chunks)` — the difference is what
+/// dedup saved. On a warm peer with an identical prior build both
+/// numbers are equal to `manifest.chunks.len()` minus the count that
+/// were already present, and only the missing-chunk bytes cross the
+/// wire.
+pub async fn push_blob_missing_chunks(
+    conn: &Connection,
+    local: &BlobStore,
+    id: &BlobId,
+) -> Result<(usize, usize)> {
+    let manifest = local
+        .load_manifest(id)
+        .await?
+        .with_context(|| format!("blob {} not present locally", id.to_hex()))?;
+    let total = manifest.chunks.len();
+    let mut uploaded = 0usize;
+    for hash in &manifest.chunks {
+        if !call_has_chunk(conn, hash).await? {
+            let bytes = local
+                .read_chunk(hash)
+                .await?
+                .with_context(|| format!("chunk {} missing locally", hash.to_hex()))?;
+            call_put_chunk(conn, hash, &bytes).await?;
+            uploaded += 1;
+        }
+    }
+    // Commit the manifest. `missing` MUST be empty by now — if the
+    // peer still reports missing chunks after we uploaded them, the
+    // most likely cause is a store crash on their side; surface as Err.
+    let still_missing = call_put_manifest(conn, &manifest).await?;
+    if !still_missing.is_empty() {
+        bail!(
+            "peer still reports {} missing chunks after we uploaded {} — retry",
+            still_missing.len(),
+            uploaded
+        );
+    }
+    Ok((uploaded, total))
+}
+
 #[cfg(test)]
 #[path = "rpc/tests.rs"]
 mod tests;

claw-store/src/cluster/rpc/tests.rs

@@ -513,6 +513,252 @@ fn method_reports_streaming_variants() {
     assert!(!Method::BlobLoadManifest.is_streaming());
     assert!(Method::BlobPutStream.is_streaming());
     assert!(Method::BlobGetStream.is_streaming());
+    assert!(!Method::HasChunk.is_streaming());
+    assert!(!Method::PutChunk.is_streaming());
+    assert!(!Method::GetChunk.is_streaming());
+    assert!(!Method::PutManifest.is_streaming());
+}
+
+#[test]
+fn phase_2d_method_byte_encoding() {
+    assert_eq!(Method::HasChunk.as_byte(), 0x09);
+    assert_eq!(Method::PutChunk.as_byte(), 0x0a);
+    assert_eq!(Method::GetChunk.as_byte(), 0x0b);
+    assert_eq!(Method::PutManifest.as_byte(), 0x0c);
+    assert_eq!(Method::from_byte(0x09), Some(Method::HasChunk));
+    assert_eq!(Method::from_byte(0x0a), Some(Method::PutChunk));
+    assert_eq!(Method::from_byte(0x0b), Some(Method::GetChunk));
+    assert_eq!(Method::from_byte(0x0c), Some(Method::PutManifest));
+}
+
+// ── Phase 2d: chunk-level RPC ────────────────────────────────────────
+
+#[tokio::test]
+async fn has_chunk_returns_ok_for_present_and_not_found_for_missing() {
+    let (_tmp, router) = router_with_blobs("solo", next_port()).await;
+    let bytes = b"chunky content";
+    let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
+    router
+        .blob_store()
+        .unwrap()
+        .put_chunk(&hash, bytes)
+        .await
+        .unwrap();
+
+    let mut req = vec![Method::HasChunk.as_byte()];
+    req.extend_from_slice(hash.as_bytes());
+    assert_eq!(dispatch(&router, &req).await, vec![STREAM_STATUS_OK]);
+
+    let mut req = vec![Method::HasChunk.as_byte()];
+    req.extend_from_slice(&[0u8; 32]);
+    assert_eq!(dispatch(&router, &req).await, vec![ErrorCode::NotFound.as_byte()]);
+}
+
+#[tokio::test]
+async fn put_chunk_stores_and_returns_status_ok() {
+    let (_tmp, router) = router_with_blobs("solo", next_port()).await;
+    let bytes = b"payload for put_chunk";
+    let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
+    let mut req = vec![Method::PutChunk.as_byte()];
+    req.extend_from_slice(hash.as_bytes());
+    req.extend_from_slice(bytes);
+    assert_eq!(dispatch(&router, &req).await, vec![STREAM_STATUS_OK]);
+    assert!(router.blob_store().unwrap().has_chunk(&hash).await.unwrap());
+}
+
+#[tokio::test]
+async fn put_chunk_rejects_hash_mismatch_over_wire() {
+    let (_tmp, router) = router_with_blobs("solo", next_port()).await;
+    let bytes = b"real";
+    let bogus = ChunkHash::from_bytes([0xffu8; 32]);
+    let mut req = vec![Method::PutChunk.as_byte()];
+    req.extend_from_slice(bogus.as_bytes());
+    req.extend_from_slice(bytes);
+    let reply = dispatch(&router, &req).await;
+    assert_eq!(reply, vec![ErrorCode::InvalidRequest.as_byte()]);
+    assert!(!router.blob_store().unwrap().has_chunk(&bogus).await.unwrap());
+}
+
+#[tokio::test]
+async fn get_chunk_returns_content_prefixed_with_status_ok() {
+    let (_tmp, router) = router_with_blobs("solo", next_port()).await;
+    let bytes = b"fetch me by hash";
+    let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
+    router
+        .blob_store()
+        .unwrap()
+        .put_chunk(&hash, bytes)
+        .await
+        .unwrap();
+
+    let mut req = vec![Method::GetChunk.as_byte()];
+    req.extend_from_slice(hash.as_bytes());
+    let reply = dispatch(&router, &req).await;
+    assert_eq!(reply[0], STREAM_STATUS_OK);
+    assert_eq!(&reply[1..], bytes);
+}
+
+#[tokio::test]
+async fn get_chunk_returns_not_found_for_missing() {
+    let (_tmp, router) = router_with_blobs("solo", next_port()).await;
+    let missing = ChunkHash::from_bytes([0u8; 32]);
+    let mut req = vec![Method::GetChunk.as_byte()];
+    req.extend_from_slice(missing.as_bytes());
+    let reply = dispatch(&router, &req).await;
+    assert_eq!(reply, vec![ErrorCode::NotFound.as_byte()]);
+}
+
+#[tokio::test]
+async fn put_manifest_reports_missing_chunks_when_incomplete() {
+    let (_tmp, router) = router_with_blobs("solo", next_port()).await;
+    let phantom = ChunkHash::from_bytes([0xaau8; 32]);
+    let manifest = BlobManifest {
+        blob_id: BlobId::from_bytes([0x11u8; 32]),
+        total_size: 42,
+        chunks: vec![phantom],
+    };
+    let mut req = vec![Method::PutManifest.as_byte()];
+    req.extend_from_slice(&serde_json::to_vec(&manifest).unwrap());
+    let reply = dispatch(&router, &req).await;
+    let decoded: PutManifestReply = serde_json::from_slice(&reply).unwrap();
+    assert_eq!(decoded.blob_id, manifest.blob_id);
+    assert_eq!(decoded.missing, vec![phantom]);
+}
+
+#[tokio::test]
+async fn put_manifest_writes_when_chunks_present() {
+    let (_tmp, router) = router_with_blobs("solo", next_port()).await;
+    let store = router.blob_store().unwrap().clone();
+    let a = b"first chunk";
+    let b = b"second chunk";
+    let hash_a = ChunkHash::from_bytes(blake3::hash(a).into());
+    let hash_b = ChunkHash::from_bytes(blake3::hash(b).into());
+    store.put_chunk(&hash_a, a).await.unwrap();
+    store.put_chunk(&hash_b, b).await.unwrap();
+
+    let manifest = BlobManifest {
+        blob_id: BlobId::from_bytes([0x22u8; 32]),
+        total_size: (a.len() + b.len()) as u64,
+        chunks: vec![hash_a, hash_b],
+    };
+    let mut req = vec![Method::PutManifest.as_byte()];
+    req.extend_from_slice(&serde_json::to_vec(&manifest).unwrap());
+    let reply = dispatch(&router, &req).await;
+    let decoded: PutManifestReply = serde_json::from_slice(&reply).unwrap();
+    assert!(decoded.missing.is_empty());
+    assert!(store.contains(&manifest.blob_id).await.unwrap());
+}
+
+#[tokio::test]
+async fn chunk_rpcs_return_not_configured_without_store() {
+    let gossip = bootstrap_gossip("solo", next_port()).await;
+    let router = RpcRouter::new(gossip, "solo".into(), "z".into());
+    for method in [
+        Method::HasChunk,
+        Method::PutChunk,
+        Method::GetChunk,
+        Method::PutManifest,
+    ] {
+        let mut req = vec![method.as_byte()];
+        req.extend_from_slice(&[0u8; 32]);
+        let reply = dispatch(&router, &req).await;
+        assert_eq!(
+            reply,
+            vec![ErrorCode::NotConfigured.as_byte()],
+            "method {method:?} should be NotConfigured"
+        );
+    }
+}
+
+#[tokio::test]
+async fn end_to_end_push_blob_missing_chunks_replicates_only_needed_bytes() {
+    // Peer A already has one of the two chunks of a blob;
+    // push_blob_missing_chunks uploads only the missing one and
+    // commits the manifest. Verified via the uploaded/total counters.
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp_a, router_a) = router_with_blobs("a", next_port()).await;
+    let store_a = router_a.blob_store().unwrap().clone();
+
+    let server = QuicServer::bind(loopback(0), id_a).unwrap();
+    let server_addr = server.local_addr().unwrap();
+    let router_srv = router_a.clone();
+    let accept_task = tokio::spawn(async move {
+        if let Some(Ok(conn)) = server.accept().await {
+            let _ = serve_connection(conn, router_srv).await;
+        }
+    });
+
+    let tmp_b = tempfile::TempDir::new().unwrap();
+    let store_b = crate::cluster::blob::BlobStore::open(tmp_b.path().to_path_buf()).unwrap();
+    let payload = {
+        let mut v = Vec::with_capacity(2 * crate::cluster::blob::CHUNK_SIZE);
+        v.extend(std::iter::repeat(0x11u8).take(crate::cluster::blob::CHUNK_SIZE));
+        v.extend(std::iter::repeat(0x22u8).take(crate::cluster::blob::CHUNK_SIZE));
+        v
+    };
+    let blob_id = store_b.put_bytes(&payload).await.unwrap();
+    let manifest_b = store_b.load_manifest(&blob_id).await.unwrap().unwrap();
+    assert_eq!(manifest_b.chunks.len(), 2);
+
+    // Pre-seed A with the FIRST chunk only.
+    let first_bytes = &payload[..crate::cluster::blob::CHUNK_SIZE];
+    store_a
+        .put_chunk(&manifest_b.chunks[0], first_bytes)
+        .await
+        .unwrap();
+
+    let client = QuicClient::new(loopback(0), id_b).unwrap();
+    let conn = client.connect(server_addr, "a").await.unwrap();
+
+    let (uploaded, total) = push_blob_missing_chunks(&conn, &store_b, &blob_id)
+        .await
+        .unwrap();
+    assert_eq!(total, 2);
+    assert_eq!(uploaded, 1, "only the missing chunk should have crossed");
+
+    let round = store_a.get_bytes(&blob_id).await.unwrap().unwrap();
+    assert_eq!(round, payload);
+
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    accept_task.abort();
+}
+
+#[tokio::test]
+async fn call_get_chunk_verifies_returned_hash() {
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp, router) = router_with_blobs("a", next_port()).await;
+    let bytes = b"chunk to fetch";
+    let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
+    router
+        .blob_store()
+        .unwrap()
+        .put_chunk(&hash, bytes)
+        .await
+        .unwrap();
+
+    let server = QuicServer::bind(loopback(0), id_a).unwrap();
+    let server_addr = server.local_addr().unwrap();
+    let router_srv = router.clone();
+    let accept_task = tokio::spawn(async move {
+        if let Some(Ok(conn)) = server.accept().await {
+            let _ = serve_connection(conn, router_srv).await;
+        }
+    });
+
+    let client = QuicClient::new(loopback(0), id_b).unwrap();
+    let conn = client.connect(server_addr, "a").await.unwrap();
+    let got = call_get_chunk(&conn, &hash).await.unwrap().unwrap();
+    assert_eq!(got, bytes);
+
+    let missing = ChunkHash::from_bytes([0u8; 32]);
+    assert!(call_get_chunk(&conn, &missing).await.unwrap().is_none());
+
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    accept_task.abort();
 }
 
 #[tokio::test]