claw-store/src/cluster/blob.rs

@@ -168,6 +168,26 @@ pub struct GcReport {
     pub bytes_reclaimed: u64,
 }
 
+/// Phase 7a (2026-07-14): report from [`BlobStore::scrub_all`].
+///
+/// A scrub walks every manifest, recomputes BLAKE3 for each referenced
+/// chunk file, and reports mismatches without touching disk state.
+/// Read-only; safe to run against a live daemon.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ScrubReport {
+    pub manifests_scanned: usize,
+    pub chunks_scanned: usize,
+    pub chunks_ok: usize,
+    pub chunks_corrupt: usize,
+    pub chunks_missing: usize,
+    /// (owning blob, chunk-hash whose file contents don't hash to that hash).
+    /// Bounded by `chunks_corrupt`; kept explicit so operators can act.
+    pub corrupt_chunks: Vec<(BlobId, ChunkHash)>,
+    /// (owning blob, chunk-hash whose file is absent from disk).
+    /// Bounded by `chunks_missing`.
+    pub missing_chunks: Vec<(BlobId, ChunkHash)>,
+}
+
 /// Content-addressed blob store rooted at a filesystem directory.
 #[derive(Debug, Clone)]
 pub struct BlobStore {
@@ -850,6 +870,116 @@ impl BlobStore {
         }
         Ok(referenced)
     }
+
+    /// Phase 7a (2026-07-14): read-only fsck for the blob store.
+    ///
+    /// For every `.manifest.json`, for every chunk it references:
+    /// * If the chunk file is absent → count as `missing`.
+    /// * If present but its BLAKE3 doesn't match the manifest's hash
+    ///   → count as `corrupt`.
+    /// * Otherwise → `ok`.
+    ///
+    /// Chunks shared across multiple manifests are counted per
+    /// reference (not per unique on-disk file) so operators see the
+    /// full blast radius: one bad chunk that 5 blobs depend on shows
+    /// up as 5 corrupt entries in `corrupt_chunks`. Cheap because we
+    /// still hash the file only once per unique chunk in memory (via
+    /// a `verified` cache in the loop).
+    ///
+    /// Never mutates disk. Safe against a live daemon: worst case a
+    /// chunk lands mid-scrub and is missed this round.
+    pub async fn scrub_all(&self) -> Result<ScrubReport> {
+        let blobs_root = self.root.join("blobs");
+        let mut report = ScrubReport {
+            manifests_scanned: 0,
+            chunks_scanned: 0,
+            chunks_ok: 0,
+            chunks_corrupt: 0,
+            chunks_missing: 0,
+            corrupt_chunks: Vec::new(),
+            missing_chunks: Vec::new(),
+        };
+        // Per-scrub cache: chunk-hash → verdict. Same chunk referenced
+        // by N manifests is hashed exactly once from disk.
+        let mut verdict: std::collections::HashMap<ChunkHash, ChunkVerdict> =
+            std::collections::HashMap::new();
+
+        let mut top = tokio::fs::read_dir(&blobs_root)
+            .await
+            .with_context(|| format!("reading {}", blobs_root.display()))?;
+        while let Some(bucket) = top.next_entry().await? {
+            if !bucket.file_type().await?.is_dir() {
+                continue;
+            }
+            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
+            while let Some(entry) = inner.next_entry().await? {
+                if !entry.file_type().await?.is_file() {
+                    continue;
+                }
+                let name = entry.file_name();
+                let name_str = match name.to_str() {
+                    Some(s) => s,
+                    None => continue,
+                };
+                if !name_str.ends_with(".manifest.json") {
+                    continue;
+                }
+                let mbytes = match tokio::fs::read(entry.path()).await {
+                    Ok(b) => b,
+                    Err(_) => continue,
+                };
+                let manifest: BlobManifest =
+                    match serde_json::from_slice(&mbytes) {
+                        Ok(m) => m,
+                        Err(_) => continue,
+                    };
+                report.manifests_scanned += 1;
+                let blob_id = manifest.blob_id;
+                for chunk in &manifest.chunks {
+                    report.chunks_scanned += 1;
+                    let v = match verdict.get(chunk) {
+                        Some(v) => *v,
+                        None => {
+                            let path = self.chunk_path(chunk);
+                            let v = match tokio::fs::read(&path).await {
+                                Err(_) => ChunkVerdict::Missing,
+                                Ok(data) => {
+                                    let got: [u8; 32] =
+                                        blake3::hash(&data).into();
+                                    if got == *chunk.as_bytes() {
+                                        ChunkVerdict::Ok
+                                    } else {
+                                        ChunkVerdict::Corrupt
+                                    }
+                                }
+                            };
+                            verdict.insert(*chunk, v);
+                            v
+                        }
+                    };
+                    match v {
+                        ChunkVerdict::Ok => report.chunks_ok += 1,
+                        ChunkVerdict::Missing => {
+                            report.chunks_missing += 1;
+                            report.missing_chunks.push((blob_id, *chunk));
+                        }
+                        ChunkVerdict::Corrupt => {
+                            report.chunks_corrupt += 1;
+                            report.corrupt_chunks.push((blob_id, *chunk));
+                        }
+                    }
+                }
+            }
+        }
+        Ok(report)
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum ChunkVerdict {
+    Ok,
+    Missing,
+    Corrupt,
 }
 
 /// Monotonic counter to disambiguate temp file names within a single
@@ -1497,6 +1627,132 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn scrub_reports_all_ok_when_store_is_healthy() {
+        // Phase 7a happy path: three unrelated blobs, all chunks intact.
+        // Scrub must scan every manifest+chunk and report zero
+        // corrupt/missing.
+        let (_tmp, store) = open_store();
+        store.put_bytes(b"alpha payload").await.unwrap();
+        store.put_bytes(b"beta payload").await.unwrap();
+        store.put_bytes(&vec![0xABu8; 4096]).await.unwrap();
+
+        let r = store.scrub_all().await.unwrap();
+        assert_eq!(r.manifests_scanned, 3);
+        assert!(r.chunks_scanned >= 3);
+        assert_eq!(r.chunks_ok, r.chunks_scanned);
+        assert_eq!(r.chunks_corrupt, 0);
+        assert_eq!(r.chunks_missing, 0);
+        assert!(r.corrupt_chunks.is_empty());
+        assert!(r.missing_chunks.is_empty());
+    }
+
+    #[tokio::test]
+    async fn scrub_detects_corrupt_chunk() {
+        // Overwrite a live chunk with different bytes. Scrub must
+        // find it AND tie it back to the owning blob id.
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(b"scrub-corrupt payload").await.unwrap();
+        let hex = id.to_hex();
+        let bucket = store.root().join("chunks").join(&hex[..2]);
+        let entries: Vec<_> = std::fs::read_dir(&bucket)
+            .unwrap()
+            .filter_map(|r| r.ok())
+            .collect();
+        assert_eq!(entries.len(), 1, "one-chunk blob for a small payload");
+        std::fs::write(entries[0].path(), b"scrub-tampered").unwrap();
+
+        let r = store.scrub_all().await.unwrap();
+        assert_eq!(r.manifests_scanned, 1);
+        assert_eq!(r.chunks_scanned, 1);
+        assert_eq!(r.chunks_corrupt, 1);
+        assert_eq!(r.chunks_ok, 0);
+        assert_eq!(r.chunks_missing, 0);
+        assert_eq!(r.corrupt_chunks.len(), 1);
+        assert_eq!(r.corrupt_chunks[0].0, id, "corrupt chunk owned by our blob");
+    }
+
+    #[tokio::test]
+    async fn scrub_detects_missing_chunk() {
+        // Delete a live chunk out from under the manifest. Scrub
+        // must count it as missing (not corrupt) and record the
+        // owning blob.
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(b"scrub-missing payload").await.unwrap();
+        let hex = id.to_hex();
+        let bucket = store.root().join("chunks").join(&hex[..2]);
+        let entries: Vec<_> = std::fs::read_dir(&bucket)
+            .unwrap()
+            .filter_map(|r| r.ok())
+            .collect();
+        assert_eq!(entries.len(), 1);
+        std::fs::remove_file(entries[0].path()).unwrap();
+
+        let r = store.scrub_all().await.unwrap();
+        assert_eq!(r.manifests_scanned, 1);
+        assert_eq!(r.chunks_scanned, 1);
+        assert_eq!(r.chunks_missing, 1);
+        assert_eq!(r.chunks_corrupt, 0);
+        assert_eq!(r.chunks_ok, 0);
+        assert_eq!(r.missing_chunks.len(), 1);
+        assert_eq!(r.missing_chunks[0].0, id);
+    }
+
+    #[tokio::test]
+    async fn scrub_dedups_shared_chunk_hashing_once() {
+        // Two manifests that share the exact same single-chunk
+        // payload → same chunk-hash on disk. Corrupt it once.
+        // Scrub must report it as corrupt in BOTH manifest contexts
+        // (2 entries in corrupt_chunks) but only hit the disk read
+        // once — enforced indirectly by the fact that both entries
+        // share the same chunk hash.
+        let (_tmp, store) = open_store();
+        let id1 = store.put_bytes(b"shared payload").await.unwrap();
+        let id2 = store.put_bytes(b"shared payload").await.unwrap();
+        assert_eq!(id1, id2, "content-addressed → identical id");
+        // But dedupe on manifest write means only one manifest.
+        // Force a second manifest reference by writing a differently-
+        // named blob whose manifest points at the same chunk.
+        let hex = id1.to_hex();
+        let bucket = store.root().join("chunks").join(&hex[..2]);
+        let chunk_files: Vec<_> = std::fs::read_dir(&bucket)
+            .unwrap()
+            .filter_map(|r| r.ok())
+            .collect();
+        assert_eq!(chunk_files.len(), 1);
+        // Fabricate a second manifest pointing at the same chunk.
+        let fake_blob_hash = blake3::hash(b"different blob id").into();
+        let fake_id = BlobId::from_bytes(fake_blob_hash);
+        let fake_hex = fake_id.to_hex();
+        let fake_bucket = store.root().join("blobs").join(&fake_hex[..2]);
+        std::fs::create_dir_all(&fake_bucket).unwrap();
+        let chunk_name = chunk_files[0].file_name();
+        let chunk_hash_hex = chunk_name.to_str().unwrap();
+        let manifest = BlobManifest {
+            blob_id: fake_id,
+            total_size: 14,
+            chunks: vec![ChunkHash::from_hex(chunk_hash_hex).unwrap()],
+        };
+        std::fs::write(
+            fake_bucket.join(format!("{fake_hex}.manifest.json")),
+            serde_json::to_vec(&manifest).unwrap(),
+        )
+        .unwrap();
+
+        // Now corrupt the single shared chunk.
+        std::fs::write(chunk_files[0].path(), b"corrupted").unwrap();
+
+        let r = store.scrub_all().await.unwrap();
+        assert_eq!(r.manifests_scanned, 2);
+        assert_eq!(r.chunks_scanned, 2, "counted per-reference");
+        assert_eq!(r.chunks_corrupt, 2, "same chunk, both refs");
+        assert_eq!(r.corrupt_chunks.len(), 2);
+        let owners: std::collections::HashSet<_> =
+            r.corrupt_chunks.iter().map(|(id, _)| *id).collect();
+        assert!(owners.contains(&id1));
+        assert!(owners.contains(&fake_id));
+    }
+
     /// Recursive count of regular files under `root`. Test helper.
     fn count_files_under(root: &Path) -> usize {
         if !root.exists() {

claw-store/src/main.rs

@@ -172,6 +172,15 @@ enum Cmd {
         #[arg(long)]
         evict_to_gb: Option<u64>,
     },
+    /// Phase 7a (2026-07-14): read-only fsck for the local blob store.
+    /// Walks every blob manifest, recomputes BLAKE3 for each chunk,
+    /// reports missing + corrupt chunks. Never mutates disk. Safe to
+    /// run against a live daemon.
+    ClusterScrub {
+        /// Print each (blob, chunk) mismatch instead of just totals.
+        #[arg(long)]
+        verbose: bool,
+    },
 }
 
 #[tokio::main]
@@ -230,6 +239,7 @@ async fn main() -> Result<()> {
             tls_dir,
         } => cmd_cluster_ping(&name, &peer, rpc_addr, &payload, tls_dir.as_deref()).await?,
         Cmd::ClusterGc { evict_to_gb } => cmd_cluster_gc(&cfg, evict_to_gb).await?,
+        Cmd::ClusterScrub { verbose } => cmd_cluster_scrub(&cfg, verbose).await?,
         Cmd::ClusterPeerStatus {
             peer,
             rpc_addr,
@@ -329,6 +339,68 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
     Ok(())
 }
 
+async fn cmd_cluster_scrub(cfg: &Config, verbose: bool) -> Result<()> {
+    use cluster::blob::BlobStore;
+
+    let root = cfg
+        .cluster
+        .as_ref()
+        .and_then(|c| c.blob_store_root.clone())
+        .context("cluster.blob_store_root not configured; nothing to scrub")?;
+    if !root.is_dir() {
+        bail!("blob_store_root {} does not exist", root.display());
+    }
+    let store = BlobStore::open(root.clone())
+        .with_context(|| format!("opening blob store at {}", root.display()))?;
+
+    let started = std::time::Instant::now();
+    let report = store
+        .scrub_all()
+        .await
+        .context("scrub_all failed")?;
+    let elapsed = started.elapsed();
+
+    println!("── clawstor cluster-scrub ──────────────────────────");
+    println!("root:              {}", root.display());
+    println!("manifests scanned: {}", report.manifests_scanned);
+    println!("chunks scanned:    {}", report.chunks_scanned);
+    println!("  ok:              {}", report.chunks_ok);
+    println!("  missing:         {}", report.chunks_missing);
+    println!("  corrupt:         {}", report.chunks_corrupt);
+    if verbose {
+        if !report.missing_chunks.is_empty() {
+            println!();
+            println!("missing chunks:");
+            for (blob, chunk) in &report.missing_chunks {
+                println!("  blob {}  chunk {}", blob, chunk.to_hex());
+            }
+        }
+        if !report.corrupt_chunks.is_empty() {
+            println!();
+            println!("corrupt chunks:");
+            for (blob, chunk) in &report.corrupt_chunks {
+                println!("  blob {}  chunk {}", blob, chunk.to_hex());
+            }
+        }
+    } else if !report.missing_chunks.is_empty() || !report.corrupt_chunks.is_empty() {
+        println!();
+        println!("re-run with --verbose to list affected (blob, chunk) pairs");
+    }
+    println!("total elapsed:     {:?}", elapsed);
+    println!("────────────────────────────────────────────────────");
+    // Non-zero exit when the store has any integrity issue so cron
+    // jobs and CI checks surface a real failure instead of a
+    // clean-looking log.
+    if report.chunks_corrupt > 0 || report.chunks_missing > 0 {
+        anyhow::bail!(
+            "scrub found {} corrupt + {} missing chunks",
+            report.chunks_corrupt,
+            report.chunks_missing
+        );
+    }
+    Ok(())
+}
+
 async fn cmd_cluster_peer_status(
     peer: &str,
     rpc_addr: SocketAddr,