claw-store/src/main.rs

@@ -172,6 +172,28 @@ enum Cmd {
         #[arg(long)]
         evict_to_gb: Option<u64>,
     },
+    /// Phase 7c (2026-07-14): fix corrupt/missing chunks by pulling
+    /// them from a peer. Runs scrub first; if nothing bad, exits
+    /// clean. Otherwise probes the peer (HasChunk) for each unique
+    /// bad chunk and pulls it (GetChunk) when the peer has it.
+    /// Bytes are re-hashed on write, so a lying peer cannot corrupt
+    /// us further. Unrecoverable chunks (peer doesn't have) are
+    /// listed in the report — operator's cue to try another peer.
+    ClusterRepair {
+        /// Peer's node name — must match the peer's cert SAN.
+        #[arg(long)]
+        peer: String,
+        /// Peer's RPC socket. Typically gossip_port + 1.
+        #[arg(long)]
+        rpc_addr: SocketAddr,
+        /// Directory holding this node's mTLS material.
+        #[arg(long)]
+        tls_dir: PathBuf,
+        /// Scrub + report what WOULD be repaired without contacting
+        /// the peer or writing anything.
+        #[arg(long)]
+        dry_run: bool,
+    },
     /// 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
@@ -240,6 +262,12 @@ async fn main() -> Result<()> {
         } => 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::ClusterRepair {
+            peer,
+            rpc_addr,
+            tls_dir,
+            dry_run,
+        } => cmd_cluster_repair(&cfg, &peer, rpc_addr, &tls_dir, dry_run).await?,
         Cmd::ClusterPeerStatus {
             peer,
             rpc_addr,
@@ -401,6 +429,129 @@ async fn cmd_cluster_scrub(cfg: &Config, verbose: bool) -> Result<()> {
     Ok(())
 }
 
+async fn cmd_cluster_repair(
+    cfg: &Config,
+    peer: &str,
+    rpc_addr: SocketAddr,
+    tls_dir: &std::path::Path,
+    dry_run: bool,
+) -> Result<()> {
+    use cluster::blob::BlobStore;
+    use cluster::rpc::{call_get_chunk, call_has_chunk};
+    use cluster::transport::{NodeIdentity, QuicClient};
+
+    let root = cfg
+        .cluster
+        .as_ref()
+        .and_then(|c| c.blob_store_root.clone())
+        .context("cluster.blob_store_root not configured; nothing to repair")?;
+    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();
+
+    // Phase 1: scrub locally to identify the target set.
+    let scrub = store
+        .scrub_all()
+        .await
+        .context("initial scrub_all failed")?;
+
+    println!("── clawstor cluster-repair ─────────────────────────");
+    println!("root:              {}", root.display());
+    println!("peer:              {} @ {}", peer, rpc_addr);
+    if dry_run {
+        println!("mode:              DRY-RUN (no peer contact, no writes)");
+    }
+    println!("scrub:");
+    println!("  manifests:       {}", scrub.manifests_scanned);
+    println!("  chunks:          {}", scrub.chunks_scanned);
+    println!("  ok:              {}", scrub.chunks_ok);
+    println!("  missing:         {}", scrub.chunks_missing);
+    println!("  corrupt:         {}", scrub.chunks_corrupt);
+
+    if scrub.chunks_missing == 0 && scrub.chunks_corrupt == 0 {
+        println!("nothing to repair.");
+        println!("total elapsed:     {:?}", started.elapsed());
+        println!("────────────────────────────────────────────────────");
+        return Ok(());
+    }
+
+    // Dedup to one entry per unique chunk-hash — scrub emits per-reference.
+    let mut unique: std::collections::HashSet<_> = std::collections::HashSet::new();
+    for (_blob, chunk) in scrub.missing_chunks.iter().chain(scrub.corrupt_chunks.iter()) {
+        unique.insert(*chunk);
+    }
+    let targets: Vec<_> = unique.into_iter().collect();
+    println!("unique bad chunks: {}", targets.len());
+
+    if dry_run {
+        println!("dry-run: skipping peer contact + writes");
+        println!("total elapsed:     {:?}", started.elapsed());
+        println!("────────────────────────────────────────────────────");
+        return Ok(());
+    }
+
+    // Phase 2: connect to the peer.
+    let identity = NodeIdentity::from_pem_dir(tls_dir)
+        .with_context(|| format!("loading identity from {}", tls_dir.display()))?;
+    let client = QuicClient::new("0.0.0.0:0".parse()?, identity)?;
+    let conn = client
+        .connect(rpc_addr, peer)
+        .await
+        .with_context(|| format!("connecting to {peer} at {rpc_addr}"))?;
+
+    // Phase 3: repair. Fetcher probes HasChunk first (cheap) so a
+    // peer that lacks the chunk is one round-trip, not a full pull
+    // attempt.
+    let report = store
+        .repair_chunks(&targets, |hash| {
+            let conn = &conn;
+            async move {
+                if !call_has_chunk(conn, &hash).await? {
+                    return Ok(None);
+                }
+                call_get_chunk(conn, &hash).await
+            }
+        })
+        .await;
+
+    println!("repair:");
+    println!("  attempted:       {}", report.attempted);
+    println!("  repaired:        {}", report.repaired);
+    println!("  unrecoverable:   {}", report.unrecoverable.len());
+    println!("  errors:          {}", report.errors.len());
+    if !report.unrecoverable.is_empty() {
+        println!();
+        println!("unrecoverable (peer doesn't have them; try another peer):");
+        for chunk in &report.unrecoverable {
+            println!("  {}", chunk.to_hex());
+        }
+    }
+    if !report.errors.is_empty() {
+        println!();
+        println!("errors:");
+        for (chunk, e) in &report.errors {
+            println!("  {}  {}", chunk.to_hex(), e);
+        }
+    }
+    println!("total elapsed:     {:?}", started.elapsed());
+    println!("────────────────────────────────────────────────────");
+
+    // Non-zero exit when we couldn't fully repair — same rationale as
+    // cluster-scrub: cron/CI should notice, not gloss over.
+    if !report.unrecoverable.is_empty() || !report.errors.is_empty() {
+        anyhow::bail!(
+            "repair incomplete: {} unrecoverable, {} errors",
+            report.unrecoverable.len(),
+            report.errors.len()
+        );
+    }
+    Ok(())
+}
+
 async fn cmd_cluster_peer_status(
     peer: &str,
     rpc_addr: SocketAddr,