claw-store/src/cluster/blob.rs

@@ -168,6 +168,19 @@ pub struct GcReport {
     pub bytes_reclaimed: u64,
 }
 
+/// Phase 7b (2026-07-14): report from [`BlobStore::repair_chunks`].
+///
+/// For each corrupt/missing chunk the caller supplied, records what
+/// happened: successfully fetched + written, fetcher returned None
+/// (no peer had it), or the fetch itself errored (network, protocol).
+#[derive(Debug, Clone, Default)]
+pub struct RepairReport {
+    pub attempted: usize,
+    pub repaired: usize,
+    pub unrecoverable: Vec<ChunkHash>,
+    pub errors: Vec<(ChunkHash, String)>,
+}
+
 /// Phase 7a (2026-07-14): report from [`BlobStore::scrub_all`].
 ///
 /// A scrub walks every manifest, recomputes BLAKE3 for each referenced
@@ -982,6 +995,72 @@ enum ChunkVerdict {
     Corrupt,
 }
 
+impl BlobStore {
+    /// Phase 7b (2026-07-14): re-fetch a batch of chunks from a
+    /// caller-supplied source and write them locally. Intended
+    /// consumer: `cluster-repair`, which calls `scrub_all` first and
+    /// hands the missing+corrupt chunks in.
+    ///
+    /// `fetch(hash)` returns:
+    /// * `Ok(Some(bytes))` — bytes for the chunk (caller may pull
+    ///   them from any peer that has it; a wrapper walking all peers
+    ///   fits here)
+    /// * `Ok(None)` — nobody has it; recorded as unrecoverable
+    /// * `Err(e)` — network/protocol failure for this chunk;
+    ///   recorded per-chunk, doesn't abort the batch
+    ///
+    /// Bytes are re-hashed by `put_chunk` before writing, so a peer
+    /// that returns wrong bytes for a hash can't corrupt us further.
+    pub async fn repair_chunks<F, Fut>(
+        &self,
+        chunks: &[ChunkHash],
+        fetch: F,
+    ) -> RepairReport
+    where
+        F: Fn(ChunkHash) -> Fut,
+        Fut: std::future::Future<Output = Result<Option<Vec<u8>>>>,
+    {
+        let mut report = RepairReport {
+            attempted: chunks.len(),
+            ..RepairReport::default()
+        };
+        // Dedup: same chunk may be listed twice by scrub (shared).
+        let mut seen = std::collections::HashSet::new();
+        for chunk in chunks {
+            if !seen.insert(*chunk) {
+                continue;
+            }
+            match fetch(*chunk).await {
+                Ok(Some(bytes)) => {
+                    // `put_chunk` uses write-if-absent, but repair is
+                    // exactly the case where a corrupt file may already
+                    // occupy the path. Unlink first (NotFound OK), then
+                    // re-put; put_chunk still re-hashes the bytes so a
+                    // wrong-answer peer can't corrupt us.
+                    let path = self.chunk_path(chunk);
+                    match tokio::fs::remove_file(&path).await {
+                        Ok(()) => {}
+                        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+                        Err(e) => {
+                            report.errors.push((*chunk, format!("remove: {e}")));
+                            continue;
+                        }
+                    }
+                    match self.put_chunk(chunk, &bytes).await {
+                        Ok(()) => report.repaired += 1,
+                        Err(e) => {
+                            report.errors.push((*chunk, format!("put_chunk: {e}")));
+                        }
+                    }
+                }
+                Ok(None) => report.unrecoverable.push(*chunk),
+                Err(e) => report.errors.push((*chunk, e.to_string())),
+            }
+        }
+        report
+    }
+}
+
 /// Monotonic counter to disambiguate temp file names within a single
 /// process. Combined with the process id, this makes tmp filenames
 /// unique across a fleet without needing `Math.random`.
@@ -1753,6 +1832,121 @@ mod tests {
         assert!(owners.contains(&fake_id));
     }
 
+    #[tokio::test]
+    async fn repair_writes_fetched_bytes_and_marks_repaired() {
+        // Phase 7b happy path: caller passed a corrupt chunk, fetcher
+        // returned real bytes → put_chunk overwrites and repair
+        // count = 1.
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(b"phase-7b repair").await.unwrap();
+        // Snapshot the manifest to learn what chunks we have.
+        let manifest = store.load_manifest(&id).await.unwrap().unwrap();
+        assert_eq!(manifest.chunks.len(), 1);
+        let chunk = manifest.chunks[0];
+
+        // Corrupt on-disk, then repair.
+        std::fs::write(store.chunk_path(&chunk), b"tampered").unwrap();
+        // Prove scrub sees it before we repair.
+        let pre = store.scrub_all().await.unwrap();
+        assert_eq!(pre.chunks_corrupt, 1);
+
+        let real_bytes = b"phase-7b repair".to_vec();
+        let bytes_for_fetcher = real_bytes.clone();
+        let report = store
+            .repair_chunks(&[chunk], |_h| {
+                let b = bytes_for_fetcher.clone();
+                async move { Ok(Some(b)) }
+            })
+            .await;
+        assert_eq!(report.attempted, 1);
+        assert_eq!(report.repaired, 1);
+        assert!(report.unrecoverable.is_empty());
+        assert!(report.errors.is_empty());
+
+        // Post-condition: scrub is clean again.
+        let post = store.scrub_all().await.unwrap();
+        assert_eq!(post.chunks_ok, 1);
+        assert_eq!(post.chunks_corrupt, 0);
+    }
+
+    #[tokio::test]
+    async fn repair_records_unrecoverable_when_fetcher_returns_none() {
+        // Fetcher says nobody has this chunk. Report must
+        // capture it as unrecoverable; no error.
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(b"unrecoverable payload").await.unwrap();
+        let manifest = store.load_manifest(&id).await.unwrap().unwrap();
+        let chunk = manifest.chunks[0];
+        std::fs::remove_file(store.chunk_path(&chunk)).unwrap();
+
+        let report = store
+            .repair_chunks(&[chunk], |_h| async { Ok(None) })
+            .await;
+        assert_eq!(report.attempted, 1);
+        assert_eq!(report.repaired, 0);
+        assert_eq!(report.unrecoverable, vec![chunk]);
+        assert!(report.errors.is_empty());
+    }
+
+    #[tokio::test]
+    async fn repair_records_error_and_continues_batch() {
+        // Two chunks, fetcher errors on one, succeeds on other.
+        // Batch must NOT abort: second chunk still repairs.
+        let (_tmp, store) = open_store();
+        let id1 = store.put_bytes(b"batch-repair alpha").await.unwrap();
+        let id2 = store.put_bytes(b"batch-repair beta").await.unwrap();
+        let m1 = store.load_manifest(&id1).await.unwrap().unwrap();
+        let m2 = store.load_manifest(&id2).await.unwrap().unwrap();
+        let c1 = m1.chunks[0];
+        let c2 = m2.chunks[0];
+        std::fs::write(store.chunk_path(&c1), b"corrupt-1").unwrap();
+        std::fs::write(store.chunk_path(&c2), b"corrupt-2").unwrap();
+
+        let bad = c1;
+        let report = store
+            .repair_chunks(&[c1, c2], |h| async move {
+                if h == bad {
+                    Err(anyhow::anyhow!("simulated network failure"))
+                } else {
+                    Ok(Some(b"batch-repair beta".to_vec()))
+                }
+            })
+            .await;
+        assert_eq!(report.attempted, 2);
+        assert_eq!(report.repaired, 1, "beta must repair despite alpha error");
+        assert_eq!(report.errors.len(), 1);
+        assert_eq!(report.errors[0].0, c1);
+        assert!(report.errors[0].1.contains("simulated network failure"));
+    }
+
+    #[tokio::test]
+    async fn repair_dedups_duplicate_chunks_in_input() {
+        // Scrub reports a shared chunk twice (once per owning blob).
+        // Repair must fetch it exactly once — otherwise we waste
+        // a peer round-trip per reference.
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(b"dedup input").await.unwrap();
+        let manifest = store.load_manifest(&id).await.unwrap().unwrap();
+        let chunk = manifest.chunks[0];
+        std::fs::remove_file(store.chunk_path(&chunk)).unwrap();
+
+        let call_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let counter = call_count.clone();
+        let report = store
+            .repair_chunks(&[chunk, chunk, chunk], move |_h| {
+                counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                async move { Ok(Some(b"dedup input".to_vec())) }
+            })
+            .await;
+        assert_eq!(report.attempted, 3);
+        assert_eq!(report.repaired, 1, "one unique chunk actually repaired");
+        assert_eq!(
+            call_count.load(std::sync::atomic::Ordering::SeqCst),
+            1,
+            "fetcher must be called exactly once"
+        );
+    }
+
     /// Recursive count of regular files under `root`. Test helper.
     fn count_files_under(root: &Path) -> usize {
         if !root.exists() {