Phase 5k: parallel-fanout chunk transfer for prewarm

Pilot 2026-07-12 measured 109 MiB/s on the sequential prewarm path — ~11% of a 10G fabric. `quinn::Connection` is cheap-Clone (internal Arc), so we can run the has→get→put pipeline per chunk in concurrent tasks under a bounded semaphore. - `prewarm_missing_chunks_between_parallel(up, down, id, concurrency)` in `rpc/client.rs`. `concurrency <= 1` degrades to the sequential path (kept for diagnostic parity). - `claw-cargo prewarm --parallel N` (default 8). Ignored with `--buffered`. Memory ceiling: 4 MiB × in-flight = 32 MiB @ 8, 128 MiB @ 32. - Uses `tokio::task::JoinSet` + `Arc<Semaphore>`; permit held for the whole per-chunk pipeline so we never over-commit. - Retry pass on `put_manifest` mismatch stays sequential — small, correctness-critical. - Errors: JoinSet drains completely + returns first task error so a mid-fanout failure doesn't leave zombie tasks. +1 test: `end_to_end_parallel_prewarm_copies_chunks_and_matches_sequential` runs 5-chunk payload with concurrency=3, verifies byte-equal restore, then reruns with concurrency=8 → 0 uploads (has_chunk dedup), then concurrency=0 → 0 uploads (sequential fallback path). 254 tests pass (+1 from previous). Pre-existing macOS failure unchanged.
2026-07-12 06:13:28 -07:00
parent a8fac47470
commit 54e9da4d62
3 changed files with 236 additions and 13 deletions
@@ -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))
+}