claw-store/src/claw_cargo.rs

@@ -46,7 +46,8 @@ use crate::cluster::build_cache::{
 };
 use crate::cluster::client_config::{ClientConfig, ResolvedClientConfig};
 use crate::cluster::rpc::{
-    call_blob_get_stream, call_blob_put_stream, call_blob_stat, call_delete_tag, call_get_metrics, prewarm_missing_chunks_between,
+    call_blob_get_stream, call_blob_put_stream, call_blob_stat, call_delete_tag, call_get_metrics,
+    prewarm_missing_chunks_between_parallel,
     call_get_ref, call_get_tag, call_list_tags, call_put_ref, call_put_tag,
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
@@ -129,6 +130,17 @@ struct PrewarmArgs {
     /// 4 MiB chunk in RAM at a time).
     #[arg(long, default_value_t = false)]
     buffered: bool,
+    /// Phase 5k: concurrent chunk transfers on the streaming path.
+    /// Pilot 2026-07-12 measured 109 MiB/s sequential over 10G (~11%
+    /// of link cap); parallelism pushes toward the ceiling.
+    ///
+    /// Memory: 4 MiB × in-flight requests. `parallel = 8` → 32 MiB;
+    /// `parallel = 32` → 128 MiB.
+    ///
+    /// Set to `1` to force sequential (matches the pre-5k behavior).
+    /// Ignored with `--buffered`.
+    #[arg(long, default_value_t = 8)]
+    parallel: usize,
 }
 
 #[derive(clap::Args, Debug, Clone)]
@@ -896,15 +908,18 @@ async fn cmd_prewarm(args: PrewarmArgs) -> Result<()> {
         // with the streaming path's semantics.
         (stat.chunk_count as usize, bytes_len, "buffered")
     } else {
-        // Phase 5h: chunk-by-chunk streaming. Memory ceiling = one
+        // Phase 5h + 5k: chunk-by-chunk streaming with configurable
-        // chunk (4 MiB). Also gets us free dedup — chunks already
+        // fanout. Memory ceiling = 4 MiB × in-flight requests. Also
-        // present downstream (from a prior warm build) are skipped.
+        // gets us free dedup — chunks already present downstream
-        let (uploaded, total) =
+        // (from a prior warm build) are skipped.
-            prewarm_missing_chunks_between(&up_conn, &down_conn, &blob_id)
+        let (uploaded, total) = prewarm_missing_chunks_between_parallel(
-                .await
+            &up_conn,
-                .with_context(|| {
+            &down_conn,
-                    format!("streaming prewarm of blob {}", blob_id.to_hex())
+            &blob_id,
-                })?;
+            args.parallel,
+        )
+        .await
+        .with_context(|| format!("streaming prewarm of blob {}", blob_id.to_hex()))?;
         // Bytes actually transferred = uploaded chunks × chunk size,
         // capped at total_size for the tail chunk. Rough estimate; a
         // true byte count would require the router to report actual
@@ -912,13 +927,19 @@ async fn cmd_prewarm(args: PrewarmArgs) -> Result<()> {
         let approx_bytes = (uploaded as u64) * CHUNK_SIZE as u64;
         let approx_bytes = approx_bytes.min(stat.total_size);
         tracing::info!(
-            "prewarm: streamed {}/{} chunks ({} dedup skipped) in {:?}",
+            "prewarm: streamed {}/{} chunks ({} dedup skipped) in {:?} parallel={}",
             uploaded,
             total,
             total.saturating_sub(uploaded),
-            transfer_started.elapsed()
+            transfer_started.elapsed(),
+            args.parallel
         );
-        (uploaded, approx_bytes, "streaming")
+        let label = if args.parallel <= 1 {
+            "streaming"
+        } else {
+            "streaming+parallel"
+        };
+        (uploaded, approx_bytes, label)
     };
     let transfer_elapsed = transfer_started.elapsed();
 

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

@@ -676,3 +676,106 @@ pub async fn prewarm_missing_chunks_between(
     }
     Ok((uploaded, total))
 }
+
+/// Phase 5k: parallel-fanout variant of [`prewarm_missing_chunks_between`].
+///
+/// Runs the has→get→put pipeline for each chunk concurrently, bounded
+/// by `concurrency`. Pilot 2026-07-12 measured **109 MiB/s** on the
+/// sequential path — ~11% of a 10G link. Parallelism pushes toward
+/// the link cap; on the same clawverse workload (249 chunks) we
+/// expect a several-× speedup with `concurrency = 8`.
+///
+/// `concurrency = 0` or `1` degrades to the sequential path.
+///
+/// Memory: one 4 MiB chunk buffer × in-flight requests. `concurrency
+/// = 8` → 32 MiB peak; `concurrency = 32` → 128 MiB.
+///
+/// `quinn::Connection` is `Clone` (internal `Arc`) so we can share it
+/// across the spawned tasks without wrapping in an outer Arc.
+pub async fn prewarm_missing_chunks_between_parallel(
+    upstream: &Connection,
+    downstream: &Connection,
+    id: &BlobId,
+    concurrency: usize,
+) -> Result<(usize, usize)> {
+    // Fall through to the sequential path when parallelism disabled.
+    if concurrency <= 1 {
+        return prewarm_missing_chunks_between(upstream, downstream, id).await;
+    }
+    let manifest = call_blob_load_manifest(upstream, id)
+        .await?
+        .with_context(|| format!("upstream missing blob {}", id.to_hex()))?;
+    let total = manifest.chunks.len();
+
+    let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency));
+    let mut set = tokio::task::JoinSet::new();
+    for hash in manifest.chunks.iter().copied() {
+        let permit = sem
+            .clone()
+            .acquire_owned()
+            .await
+            .context("acquiring prewarm semaphore permit")?;
+        let up = upstream.clone();
+        let down = downstream.clone();
+        set.spawn(async move {
+            // Permit held for the whole pipeline — released on drop.
+            let _permit = permit;
+            if call_has_chunk(&down, &hash).await? {
+                return Ok::<bool, anyhow::Error>(false);
+            }
+            let bytes = call_get_chunk(&up, &hash).await?.with_context(|| {
+                format!(
+                    "upstream manifest referenced chunk {} but GetChunk returned NotFound",
+                    hash.to_hex()
+                )
+            })?;
+            call_put_chunk(&down, &hash, &bytes).await?;
+            Ok(true)
+        });
+    }
+
+    let mut uploaded = 0usize;
+    let mut first_err: Option<anyhow::Error> = None;
+    while let Some(join) = set.join_next().await {
+        match join {
+            Ok(Ok(pushed)) => {
+                if pushed {
+                    uploaded += 1;
+                }
+            }
+            Ok(Err(e)) => {
+                if first_err.is_none() {
+                    first_err = Some(e);
+                }
+            }
+            Err(join_err) => {
+                if first_err.is_none() {
+                    first_err = Some(anyhow::Error::from(join_err));
+                }
+            }
+        }
+    }
+    if let Some(e) = first_err {
+        return Err(e).context("prewarm chunk task failed");
+    }
+
+    // Commit + one retry pass — same shape as the sequential variant.
+    let still_missing = call_put_manifest(downstream, &manifest).await?;
+    if !still_missing.is_empty() {
+        for hash in &still_missing {
+            let bytes = call_get_chunk(upstream, hash)
+                .await?
+                .with_context(|| format!("retry fetch of chunk {} failed", hash.to_hex()))?;
+            call_put_chunk(downstream, hash, &bytes).await?;
+            uploaded += 1;
+        }
+        let final_missing = call_put_manifest(downstream, &manifest).await?;
+        if !final_missing.is_empty() {
+            bail!(
+                "downstream still missing {} chunks after retry; storage may be failing",
+                final_missing.len()
+            );
+        }
+    }
+    Ok((uploaded, total))
+}

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

@@ -1105,3 +1105,102 @@ async fn streaming_prewarm_skips_chunks_already_present_downstream() {
     acc_a.abort();
     acc_c.abort();
 }
+
+#[tokio::test]
+async fn end_to_end_parallel_prewarm_copies_chunks_and_matches_sequential() {
+    // Phase 5k: parallel prewarm variant produces the same downstream
+    // state as the sequential path. Also proves the JoinSet fanout
+    // doesn't lose or duplicate chunks.
+    use crate::cluster::blob::CHUNK_SIZE;
+
+    let (id_a, id_b_for_a) = NodeIdentity::generate_test_pair("a", "b_a").unwrap();
+    let (id_c, id_b_for_c) = NodeIdentity::generate_test_pair("c", "b_c").unwrap();
+
+    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
+    let (_tmp_c, router_c) = router_with_full_stack("c", next_port()).await;
+
+    // 5 chunks so parallelism (concurrency=3) actually queues work.
+    let payload: Vec<u8> = (0..(4 * CHUNK_SIZE + CHUNK_SIZE / 3))
+        .map(|i| ((i * 7) % 251) as u8)
+        .collect();
+    let blob_id = router_a
+        .blob_store()
+        .unwrap()
+        .put_bytes(&payload)
+        .await
+        .unwrap();
+    let manifest = router_a
+        .blob_store()
+        .unwrap()
+        .load_manifest(&blob_id)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(manifest.chunks.len(), 5, "expected 5 chunks in payload");
+
+    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
+    let addr_a = server_a.local_addr().unwrap();
+    let ra = router_a.clone();
+    let acc_a = tokio::spawn(async move {
+        while let Some(Ok(conn)) = server_a.accept().await {
+            let r = ra.clone();
+            tokio::spawn(async move {
+                let _ = serve_connection(conn, r).await;
+            });
+        }
+    });
+    let server_c = QuicServer::bind(loopback(0), id_c).unwrap();
+    let addr_c = server_c.local_addr().unwrap();
+    let rc = router_c.clone();
+    let acc_c = tokio::spawn(async move {
+        while let Some(Ok(conn)) = server_c.accept().await {
+            let r = rc.clone();
+            tokio::spawn(async move {
+                let _ = serve_connection(conn, r).await;
+            });
+        }
+    });
+
+    let up_client = QuicClient::new(loopback(0), id_b_for_a).unwrap();
+    let up_conn = up_client.connect(addr_a, "a").await.unwrap();
+    let down_client = QuicClient::new(loopback(0), id_b_for_c).unwrap();
+    let down_conn = down_client.connect(addr_c, "c").await.unwrap();
+
+    let (uploaded, total) =
+        prewarm_missing_chunks_between_parallel(&up_conn, &down_conn, &blob_id, 3)
+            .await
+            .unwrap();
+    assert_eq!(total, 5);
+    assert_eq!(uploaded, 5, "cold downstream must receive every chunk");
+
+    // Round-trip proves manifest committed AND chunks landed.
+    let round = router_c
+        .blob_store()
+        .unwrap()
+        .get_bytes(&blob_id)
+        .await
+        .unwrap();
+    assert_eq!(round.as_deref(), Some(payload.as_slice()));
+
+    // Idempotency: re-running with concurrency=8 uploads 0 (full dedup).
+    let (uploaded2, _) =
+        prewarm_missing_chunks_between_parallel(&up_conn, &down_conn, &blob_id, 8)
+            .await
+            .unwrap();
+    assert_eq!(uploaded2, 0, "re-run should be a no-op via has_chunk dedup");
+
+    // concurrency=0 falls through to sequential.
+    let (uploaded3, _) =
+        prewarm_missing_chunks_between_parallel(&up_conn, &down_conn, &blob_id, 0)
+            .await
+            .unwrap();
+    assert_eq!(uploaded3, 0, "sequential fallback should also see full dedup");
+
+    up_conn.close(quinn::VarInt::from_u32(0), b"done");
+    down_conn.close(quinn::VarInt::from_u32(0), b"done");
+    up_client.shutdown().await;
+    down_client.shutdown().await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    acc_a.abort();
+    acc_c.abort();
+}