Phase 5e: prefetch --pin <tag> #14

Merged
osobh merged 1 commits from phase-5e-prefetch-pin into main 2026-07-12 11:15:36 +00:00
2 changed files with 144 additions and 8 deletions
+76 -8
View File
@@ -68,8 +68,12 @@ enum Cmd {
/// cargo build + capture + upload.
Build(BuildArgs),
/// Download + restore the cached target dir into `target/<profile>`
/// without running cargo. Prints hit/miss + bytes.
Prefetch(PeerArgs),
/// without running cargo. Prints hit/miss + bytes. Accepts either a
/// fingerprint-based lookup (the default, computed from workspace
/// state) or a `--pin <tag>` override that resolves the tag to a
/// BlobId via `GetTag` — useful for restoring an old cache by
/// name without needing the source tree to match.
Prefetch(PrefetchArgs),
/// Print the fingerprint + peer cache state, no build, no download.
Status(PeerArgs),
/// Print the fingerprint only. Local, no network.
@@ -138,6 +142,21 @@ struct LocalArgs {
workspace: Option<PathBuf>,
}
/// `prefetch` extends the peer args with an optional tag override.
#[derive(clap::Args, Debug, Clone)]
struct PrefetchArgs {
#[command(flatten)]
peer: PeerArgs,
/// Optional: skip fingerprint compute and resolve the given tag
/// via `GetTag` to find the BlobId to download. When set,
/// `profile` is still used to pick the destination directory
/// (`target/<profile>`), but `features` and workspace state are
/// otherwise irrelevant. Useful for restoring an old cache
/// (e.g. `--pin clawverse:main:2026-07-12`) into a fresh checkout.
#[arg(long)]
pin: Option<String>,
}
/// `build` gets peer args + a couple extras.
#[derive(clap::Args, Debug, Clone)]
struct BuildArgs {
@@ -300,12 +319,29 @@ async fn cmd_status(args: PeerArgs) -> Result<()> {
// ── prefetch ─────────────────────────────────────────────────────────
async fn cmd_prefetch(args: PeerArgs) -> Result<()> {
let (workspace, resolved, fp) = setup_peer(&args)?;
async fn cmd_prefetch(args: PrefetchArgs) -> Result<()> {
let (workspace, resolved, fp) = setup_peer(&args.peer)?;
let target_dir = workspace.join("target").join(&resolved.profile);
let (client, conn) = connect_peer(&resolved).await?;
let outcome = match peer_lookup(&conn, &fp).await? {
// Two lookup paths depending on --pin:
// * pin=None → fingerprint → ref → BlobId (Phase 5b default)
// * pin=Some → tag → BlobId (Phase 5d/5e — restore a named cache)
let lookup = match &args.pin {
Some(tag) => match resolve_pin(&conn, tag).await? {
Some(pair) => Some(pair),
None => {
conn.close(quinn::VarInt::from_u32(0), b"done");
client.shutdown().await;
println!("no such tag: {}", tag);
return Ok(());
}
},
None => peer_lookup(&conn, &fp).await?,
};
let outcome = match lookup {
Some((blob_id, stat)) => {
println!(
"cache HIT — downloading {} bytes ({} chunks) to {}",
@@ -317,7 +353,7 @@ async fn cmd_prefetch(args: PeerArgs) -> Result<()> {
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 {
println!("cache: MISS (ref pointed at a missing blob)");
println!("cache: MISS (ref/tag pointed at a missing blob)");
PrefetchOutcome::Miss
} else {
std::fs::create_dir_all(&target_dir)
@@ -346,7 +382,10 @@ async fn cmd_prefetch(args: PeerArgs) -> Result<()> {
} => {
println!();
println!("── claw-cargo prefetch ─────────────────────────────");
println!("fingerprint: {}", fp);
match &args.pin {
Some(tag) => println!("source: --pin {}", tag),
None => println!("fingerprint: {}", fp),
}
println!("blob: {}", blob_id);
println!("downloaded: {} bytes in {:?}", bytes, elapsed);
println!("restored to: {}", target_dir.display());
@@ -354,7 +393,10 @@ async fn cmd_prefetch(args: PeerArgs) -> Result<()> {
}
PrefetchOutcome::Miss => {
println!();
println!("fingerprint: {} — no cache to prefetch", fp);
match &args.pin {
Some(tag) => println!("--pin {} — no cache to prefetch", tag),
None => println!("fingerprint: {} — no cache to prefetch", fp),
}
}
}
Ok(())
@@ -369,6 +411,32 @@ enum PrefetchOutcome {
Miss,
}
/// Look up a `--pin` tag → BlobId → BlobStat. Returns `Ok(None)` when
/// no such tag exists on the peer; errors when the tag is set but its
/// blob has been garbage-collected (mirroring the same-signature
/// `peer_lookup` behaviour so callers can treat both as "no cache").
async fn resolve_pin(
conn: &quinn::Connection,
tag: &str,
) -> Result<Option<(BlobId, crate::cluster::blob::BlobStat)>> {
let value = match call_get_tag(conn, tag).await? {
Some(v) => v,
None => return Ok(None),
};
let blob_id = BlobId::from_bytes(value);
match call_blob_stat(conn, &blob_id).await? {
Some(stat) => Ok(Some((blob_id, stat))),
None => {
tracing::warn!(
"tag {} points at blob {} but peer has no such blob; treating as miss",
tag,
blob_id
);
Ok(None)
}
}
}
// ── build ────────────────────────────────────────────────────────────
async fn cmd_build(args: BuildArgs) -> Result<()> {
+68
View File
@@ -976,6 +976,74 @@ async fn list_tags_returns_json_sorted() {
assert_eq!(decoded[0].decode_value().unwrap(), [1u8; 32]);
}
#[tokio::test]
async fn end_to_end_tag_resolve_and_stream_restore_over_real_quic() {
// Phase 5e: the pipeline `claw-cargo prefetch --pin <tag>` runs
// internally — put a blob, publish a tag pointing at it, then
// GetTag → BlobStat → BlobGetStream to reassemble the content
// byte-equal on the other side.
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_full_stack("a", next_port()).await;
// Seed the blob store with a synthetic "captured target dir".
let payload: Vec<u8> = (0..2 * 1024 * 1024).map(|i| (i % 251) as u8).collect();
let blob_id = router
.blob_store()
.unwrap()
.put_bytes(&payload)
.await
.unwrap();
// Publish a tag pointing at that blob.
router
.tag_store()
.unwrap()
.put("clawverse:main:latest", blob_id.as_bytes())
.await
.unwrap();
let server = QuicServer::bind(loopback(0), id_a).unwrap();
let server_addr = server.local_addr().unwrap();
let router_srv = router.clone();
let accept_task = tokio::spawn(async move {
if let Some(Ok(conn)) = server.accept().await {
let _ = serve_connection(conn, router_srv).await;
}
});
let client = QuicClient::new(loopback(0), id_b).unwrap();
let conn = client.connect(server_addr, "a").await.unwrap();
// Client side of `prefetch --pin <tag>`:
// 1. GetTag → BlobId bytes
// 2. BlobStat → confirm blob exists + size
// 3. BlobGetStream → download into buffer
let tag_value = call_get_tag(&conn, "clawverse:main:latest")
.await
.unwrap()
.expect("tag set");
assert_eq!(tag_value, *blob_id.as_bytes());
let resolved_id = BlobId::from_bytes(tag_value);
let stat = call_blob_stat(&conn, &resolved_id).await.unwrap().unwrap();
assert_eq!(stat.total_size, payload.len() as u64);
let mut sink: Vec<u8> = Vec::new();
let ok = call_blob_get_stream(&conn, &resolved_id, &mut sink)
.await
.unwrap();
assert!(ok);
assert_eq!(sink.len(), payload.len());
assert_eq!(sink, payload, "reassembled bytes match source");
// Missing-tag path: prefetch --pin never-set-name reports None.
let missing = call_get_tag(&conn, "never-set").await.unwrap();
assert!(missing.is_none());
conn.close(quinn::VarInt::from_u32(0), b"done");
client.shutdown().await;
tokio::time::sleep(Duration::from_millis(50)).await;
accept_task.abort();
}
#[tokio::test]
async fn end_to_end_pin_lookup_delete_over_real_quic() {
// Full flow: publish a tag → look it up → list → delete → confirm gone.