claw-store/src/claw_cargo.rs

@@ -46,8 +46,9 @@ 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,
-    call_peer_status, prewarm_missing_chunks_between_parallel,
+    call_blob_get_parallel, call_blob_get_stream, call_blob_put_stream, call_blob_stat,
+    call_delete_tag, call_get_metrics, call_peer_status,
+    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};
@@ -219,6 +220,13 @@ struct BuildArgs {
     /// Skip capture + upload on a miss. Useful for read-only cache use.
     #[arg(long)]
     no_upload: bool,
+    /// Field finding 2026-07-12: parallel chunk fetch on cache HIT.
+    /// Sequential BlobGetStream restore of 947 MiB on Pi 5 took 18s
+    /// (~53 MiB/s); parallel chunk fetches stack their per-stream
+    /// throughputs. Set to 1 to force sequential (matches pre-fix
+    /// behavior).
+    #[arg(long, default_value_t = 8)]
+    parallel_restore: usize,
     /// Extra args passed verbatim to `cargo build` (after `--`).
     #[arg(last = true)]
     cargo_args: Vec<String>,
@@ -558,10 +566,29 @@ async fn cmd_build(args: BuildArgs) -> Result<()> {
     let mut outcome = CacheOutcome::Miss;
     match peer_lookup(&conn, &fp).await? {
         Some((blob_id, stat)) => {
-            tracing::info!("cache HIT — blob {} ({} bytes)", blob_id, stat.total_size);
-            let mut buf = Vec::with_capacity(stat.total_size as usize);
-            let ok = call_blob_get_stream(&conn, &blob_id, &mut buf).await?;
-            if ok {
+            tracing::info!(
+                "cache HIT — blob {} ({} bytes, parallel={})",
+                blob_id,
+                stat.total_size,
+                args.parallel_restore
+            );
+            // Field finding 2026-07-12: single-stream BlobGetStream
+            // capped Pi restore at ~53 MiB/s. Parallel chunk fetches
+            // let per-stream throughputs stack. `parallel_restore <= 1`
+            // falls back to the sequential BlobGetStream path (kept
+            // for diagnostic parity).
+            let buf = if args.parallel_restore <= 1 {
+                let mut b = Vec::with_capacity(stat.total_size as usize);
+                let ok = call_blob_get_stream(&conn, &blob_id, &mut b).await?;
+                if ok {
+                    Some(b)
+                } else {
+                    None
+                }
+            } else {
+                call_blob_get_parallel(&conn, &blob_id, args.parallel_restore).await?
+            };
+            if let Some(buf) = buf {
                 std::fs::create_dir_all(&target_dir).with_context(|| {
                     format!("creating {}", target_dir.display())
                 })?;

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

@@ -677,6 +677,95 @@ pub async fn prewarm_missing_chunks_between(
     Ok((uploaded, total))
 }
 
+/// Field finding 2026-07-12 (Pi restore = 18s single-stream): fetch a
+/// blob by pulling its chunks in parallel and reassembling in memory.
+/// Faster than `call_blob_get_stream` on connections with per-stream
+/// throughput ceilings — parallel streams stack their contributions.
+///
+/// Flow:
+///   1. `LoadManifest` upstream (small).
+///   2. Spawn N concurrent `GetChunk` tasks bounded by a semaphore.
+///   3. Assemble the results in manifest order into a `Vec<u8>` sized
+///      to `manifest.total_size`.
+///
+/// `concurrency <= 1` degrades to sequential (matches
+/// `call_blob_get_stream` semantics but keeps the code path uniform).
+/// Memory ceiling: `total_size + 4 MiB × in-flight` — dominated by
+/// the reassembled blob buffer itself.
+pub async fn call_blob_get_parallel(
+    conn: &Connection,
+    id: &BlobId,
+    concurrency: usize,
+) -> Result<Option<Vec<u8>>> {
+    let manifest = match call_blob_load_manifest(conn, id).await? {
+        Some(m) => m,
+        None => return Ok(None),
+    };
+    let concurrency = concurrency.max(1);
+    let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency));
+    let mut set = tokio::task::JoinSet::new();
+    for (idx, hash) in manifest.chunks.iter().copied().enumerate() {
+        let permit = sem
+            .clone()
+            .acquire_owned()
+            .await
+            .context("acquiring get_parallel semaphore permit")?;
+        let conn = conn.clone();
+        set.spawn(async move {
+            let _permit = permit;
+            let bytes = call_get_chunk(&conn, &hash).await?.with_context(|| {
+                format!(
+                    "manifest referenced chunk {} but GetChunk returned NotFound",
+                    hash.to_hex()
+                )
+            })?;
+            Ok::<(usize, Vec<u8>), anyhow::Error>((idx, bytes))
+        });
+    }
+    // Assemble in manifest order. Pre-size the outer vec so we can
+    // slot each chunk's bytes at the right offset without copying.
+    let mut out = vec![0u8; manifest.total_size as usize];
+    // Chunk boundaries: chunk i starts at i * CHUNK_SIZE.
+    let chunk_size = crate::cluster::blob::CHUNK_SIZE;
+    let mut first_err: Option<anyhow::Error> = None;
+    while let Some(join) = set.join_next().await {
+        match join {
+            Ok(Ok((idx, bytes))) => {
+                let start = idx * chunk_size;
+                let end = start + bytes.len();
+                if end > out.len() {
+                    if first_err.is_none() {
+                        first_err = Some(anyhow::anyhow!(
+                            "chunk {} at idx {} would overflow reassembly buffer \
+                             (end {}, total_size {})",
+                            manifest.chunks[idx].to_hex(),
+                            idx,
+                            end,
+                            manifest.total_size
+                        ));
+                    }
+                    continue;
+                }
+                out[start..end].copy_from_slice(&bytes);
+            }
+            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("parallel chunk fetch failed");
+    }
+    Ok(Some(out))
+}
+
 /// Phase 5k: parallel-fanout variant of [`prewarm_missing_chunks_between`].
 ///
 /// Runs the has→get→put pipeline for each chunk concurrently, bounded

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

@@ -1204,3 +1204,72 @@ async fn end_to_end_parallel_prewarm_copies_chunks_and_matches_sequential() {
     acc_a.abort();
     acc_c.abort();
 }
+
+#[tokio::test]
+async fn parallel_blob_get_reassembles_multi_chunk_blob_byte_equal() {
+    // Field finding 2026-07-12: parallel chunk fetch on restore.
+    // Verifies (1) reassembly byte-equals a sequential BlobGetStream,
+    // (2) tail chunks (not full CHUNK_SIZE) land at the right offset,
+    // (3) NotFound path returns None.
+    use crate::cluster::blob::CHUNK_SIZE;
+
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
+
+    let payload: Vec<u8> = (0..(3 * CHUNK_SIZE + CHUNK_SIZE / 5))
+        .map(|i| ((i * 13) % 251) as u8)
+        .collect();
+    let blob_id = router_a
+        .blob_store()
+        .unwrap()
+        .put_bytes(&payload)
+        .await
+        .unwrap();
+
+    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 client = QuicClient::new(loopback(0), id_b).unwrap();
+    let conn = client.connect(addr_a, "a").await.unwrap();
+
+    // Sequential reference: BlobGetStream.
+    let mut seq = Vec::new();
+    let ok = call_blob_get_stream(&conn, &blob_id, &mut seq).await.unwrap();
+    assert!(ok);
+    assert_eq!(seq, payload, "sequential fetch must be byte-equal to source");
+
+    // Parallel with concurrency = 4.
+    let par = call_blob_get_parallel(&conn, &blob_id, 4)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(par, payload, "parallel fetch must match sequential");
+    assert_eq!(par, seq, "parallel and sequential must agree");
+
+    // concurrency = 1 falls through to still-parallel (with just one
+    // in-flight) but must still be correct.
+    let ser_via_par = call_blob_get_parallel(&conn, &blob_id, 1)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(ser_via_par, payload);
+
+    // Unknown blob → None.
+    let missing = crate::cluster::blob::BlobId::from_bytes([0u8; 32]);
+    let none = call_blob_get_parallel(&conn, &missing, 4).await.unwrap();
+    assert!(none.is_none(), "NotFound must surface as None");
+
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    acc_a.abort();
+}

claw-store/src/manifest.rs

@@ -127,7 +127,31 @@ impl Manifest {
         }
     }
 
+    /// Field finding 2026-07-12 (Pi deploy on vision-02): the old
+    /// hardcoded `/var/lib/claw-store/projects.toml` broke under
+    /// `ProtectSystem=strict` in the user-mode systemd unit because
+    /// /var is read-only. Follow the XDG Base Directory spec so a
+    /// user-mode install writes under `$HOME/.local/state`, and only
+    /// root installs land in `/var/lib`.
+    ///
+    /// Precedence:
+    ///   1. `$XDG_STATE_HOME/claw-store/projects.toml` (per spec)
+    ///   2. `$HOME/.local/state/claw-store/projects.toml` (XDG default)
+    ///   3. `/var/lib/claw-store/projects.toml` (system fallback)
     pub fn default_path() -> PathBuf {
+        if let Ok(xdg) = std::env::var("XDG_STATE_HOME") {
+            if !xdg.is_empty() {
+                return PathBuf::from(xdg)
+                    .join("claw-store")
+                    .join("projects.toml");
+            }
+        }
+        if let Ok(home) = std::env::var("HOME") {
+            if !home.is_empty() {
+                return PathBuf::from(home)
+                    .join(".local/state/claw-store/projects.toml");
+            }
+        }
         PathBuf::from("/var/lib/claw-store/projects.toml")
     }
 }
@@ -207,6 +231,50 @@ mod tests {
     use super::*;
     use tempfile::NamedTempFile;
 
+    #[test]
+    fn default_path_honours_xdg_state_home() {
+        // Field finding 2026-07-12: precedence XDG_STATE_HOME →
+        // $HOME/.local/state → /var/lib. Guard: an env-set XDG wins
+        // over HOME; empty XDG is treated as unset.
+        // NOTE: env mutation is process-global, so this test does its
+        // own setup/teardown and doesn't run in parallel with another
+        // that touches the same vars.
+        let saved_xdg = std::env::var("XDG_STATE_HOME").ok();
+        let saved_home = std::env::var("HOME").ok();
+
+        // Case 1: XDG_STATE_HOME wins.
+        std::env::set_var("XDG_STATE_HOME", "/tmp/xdg-fake");
+        std::env::set_var("HOME", "/tmp/home-fake");
+        assert_eq!(
+            Manifest::default_path(),
+            PathBuf::from("/tmp/xdg-fake/claw-store/projects.toml")
+        );
+
+        // Case 2: XDG unset → HOME/.local/state.
+        std::env::remove_var("XDG_STATE_HOME");
+        assert_eq!(
+            Manifest::default_path(),
+            PathBuf::from("/tmp/home-fake/.local/state/claw-store/projects.toml")
+        );
+
+        // Case 3: empty XDG treated as unset.
+        std::env::set_var("XDG_STATE_HOME", "");
+        assert_eq!(
+            Manifest::default_path(),
+            PathBuf::from("/tmp/home-fake/.local/state/claw-store/projects.toml")
+        );
+
+        // Restore.
+        match saved_xdg {
+            Some(v) => std::env::set_var("XDG_STATE_HOME", v),
+            None => std::env::remove_var("XDG_STATE_HOME"),
+        }
+        match saved_home {
+            Some(v) => std::env::set_var("HOME", v),
+            None => std::env::remove_var("HOME"),
+        }
+    }
+
     #[test]
     fn test_roundtrip_manifest() {
         let mut m = Manifest::default();