Pi deploy follow-ups: XDG default_path + parallel restore
Two fixes surfaced by the vision-02 Pi 5 measurement: ## XDG default_path `Manifest::default_path` was hardcoded to `/var/lib/claw-store/projects.toml`. That path is read-only under the user-mode systemd unit's `ProtectSystem=strict`, and creating it needs root — awful for a runner install. Precedence, matching XDG Base Directory: 1. `$XDG_STATE_HOME/claw-store/projects.toml` 2. `$HOME/.local/state/claw-store/projects.toml` 3. `/var/lib/claw-store/projects.toml` (system fallback) User-mode installs now write in $HOME by default; system installs (root, no HOME set) still land in /var/lib. +1 test: `default_path_honours_xdg_state_home` — covers all three precedence branches. Env mutation is process-global so the test saves + restores. ## Parallel restore on cache HIT Pi restore of 947 MiB via `BlobGetStream` took ~18s (~53 MiB/s) single-stream. Per-stream throughput ceilings on the connection type cap sequential fetches; parallel chunk fetches stack their contributions. - New `call_blob_get_parallel(conn, blob_id, concurrency) -> Option<Vec<u8>>` in `rpc/client.rs`. `JoinSet` + `Semaphore`, reassembles by chunk index at manifest-known offsets so out-of-order arrival is fine. - `claw-cargo build --parallel-restore N` (default 8). `N <= 1` falls through to `BlobGetStream` for parity. - Memory: `total_size + 4 MiB × in-flight` — dominated by the reassembly buffer, not the fanout. +1 test: `parallel_blob_get_reassembles_multi_chunk_blob_byte_equal` covers roundtrip byte-equality vs BlobGetStream, tail-chunk offset, concurrency=1 correctness, and NotFound → None. 259 tests pass (+2). Pre-existing macOS failure unchanged.
This commit is contained in:
Omar Sobh
@@ -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
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user