Phase 5e: prefetch --pin <tag> #14
@@ -68,8 +68,12 @@ enum Cmd {
|
|||||||
/// cargo build + capture + upload.
|
/// cargo build + capture + upload.
|
||||||
Build(BuildArgs),
|
Build(BuildArgs),
|
||||||
/// Download + restore the cached target dir into `target/<profile>`
|
/// Download + restore the cached target dir into `target/<profile>`
|
||||||
/// without running cargo. Prints hit/miss + bytes.
|
/// without running cargo. Prints hit/miss + bytes. Accepts either a
|
||||||
Prefetch(PeerArgs),
|
/// 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.
|
/// Print the fingerprint + peer cache state, no build, no download.
|
||||||
Status(PeerArgs),
|
Status(PeerArgs),
|
||||||
/// Print the fingerprint only. Local, no network.
|
/// Print the fingerprint only. Local, no network.
|
||||||
@@ -138,6 +142,21 @@ struct LocalArgs {
|
|||||||
workspace: Option<PathBuf>,
|
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.
|
/// `build` gets peer args + a couple extras.
|
||||||
#[derive(clap::Args, Debug, Clone)]
|
#[derive(clap::Args, Debug, Clone)]
|
||||||
struct BuildArgs {
|
struct BuildArgs {
|
||||||
@@ -300,12 +319,29 @@ async fn cmd_status(args: PeerArgs) -> Result<()> {
|
|||||||
|
|
||||||
// ── prefetch ─────────────────────────────────────────────────────────
|
// ── prefetch ─────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async fn cmd_prefetch(args: PeerArgs) -> Result<()> {
|
async fn cmd_prefetch(args: PrefetchArgs) -> Result<()> {
|
||||||
let (workspace, resolved, fp) = setup_peer(&args)?;
|
let (workspace, resolved, fp) = setup_peer(&args.peer)?;
|
||||||
let target_dir = workspace.join("target").join(&resolved.profile);
|
let target_dir = workspace.join("target").join(&resolved.profile);
|
||||||
|
|
||||||
let (client, conn) = connect_peer(&resolved).await?;
|
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)) => {
|
Some((blob_id, stat)) => {
|
||||||
println!(
|
println!(
|
||||||
"cache HIT — downloading {} bytes ({} chunks) to {}",
|
"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 mut buf = Vec::with_capacity(stat.total_size as usize);
|
||||||
let ok = call_blob_get_stream(&conn, &blob_id, &mut buf).await?;
|
let ok = call_blob_get_stream(&conn, &blob_id, &mut buf).await?;
|
||||||
if !ok {
|
if !ok {
|
||||||
println!("cache: MISS (ref pointed at a missing blob)");
|
println!("cache: MISS (ref/tag pointed at a missing blob)");
|
||||||
PrefetchOutcome::Miss
|
PrefetchOutcome::Miss
|
||||||
} else {
|
} else {
|
||||||
std::fs::create_dir_all(&target_dir)
|
std::fs::create_dir_all(&target_dir)
|
||||||
@@ -346,7 +382,10 @@ async fn cmd_prefetch(args: PeerArgs) -> Result<()> {
|
|||||||
} => {
|
} => {
|
||||||
println!();
|
println!();
|
||||||
println!("── claw-cargo prefetch ─────────────────────────────");
|
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!("blob: {}", blob_id);
|
||||||
println!("downloaded: {} bytes in {:?}", bytes, elapsed);
|
println!("downloaded: {} bytes in {:?}", bytes, elapsed);
|
||||||
println!("restored to: {}", target_dir.display());
|
println!("restored to: {}", target_dir.display());
|
||||||
@@ -354,7 +393,10 @@ async fn cmd_prefetch(args: PeerArgs) -> Result<()> {
|
|||||||
}
|
}
|
||||||
PrefetchOutcome::Miss => {
|
PrefetchOutcome::Miss => {
|
||||||
println!();
|
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(())
|
Ok(())
|
||||||
@@ -369,6 +411,32 @@ enum PrefetchOutcome {
|
|||||||
Miss,
|
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 ────────────────────────────────────────────────────────────
|
// ── build ────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async fn cmd_build(args: BuildArgs) -> Result<()> {
|
async fn cmd_build(args: BuildArgs) -> Result<()> {
|
||||||
|
|||||||
@@ -976,6 +976,74 @@ async fn list_tags_returns_json_sorted() {
|
|||||||
assert_eq!(decoded[0].decode_value().unwrap(), [1u8; 32]);
|
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]
|
#[tokio::test]
|
||||||
async fn end_to_end_pin_lookup_delete_over_real_quic() {
|
async fn end_to_end_pin_lookup_delete_over_real_quic() {
|
||||||
// Full flow: publish a tag → look it up → list → delete → confirm gone.
|
// Full flow: publish a tag → look it up → list → delete → confirm gone.
|
||||||
|
|||||||
Reference in New Issue
Block a user