Merge pull request 'Phase 5b: KV refs + claw-cargo CLI (killer feature, live)' (#11) from phase-5b-claw-cargo into main

Reviewed-on: #11

claw-store/Cargo.toml

@@ -7,6 +7,13 @@ edition = "2021"
 name = "claw-store"
 path = "src/main.rs"
 
+# Phase 5b: fingerprint-keyed cargo build cache CLI. Wraps `cargo build`
+# with a peer-cache lookup: hit → download+restore, miss → build+capture+upload.
+# Uses the same identity/config surface as claw-store daemon.
+[[bin]]
+name = "claw-cargo"
+path = "src/claw_cargo.rs"
+
 [dependencies]
 clap       = { version = "4", features = ["derive"] }
 serde      = { version = "1", features = ["derive"] }

claw-store/src/claw_cargo.rs

@@ -0,0 +1,319 @@
+//! `claw-cargo` — fingerprint-keyed cargo build cache (Phase 5b).
+//!
+//! Wraps `cargo build` with a peer-cache lookup. Flow:
+//!
+//! ```text
+//! 1. compute_workspace_fingerprint(workspace, profile, features)
+//! 2. QUIC + mTLS connect to peer
+//! 3. GetRef(fingerprint) → BlobId?
+//!    HIT: BlobGetStream → restore_target → cargo build (fast, just workspace)
+//!    MISS: cargo build (full) → capture_target → BlobPutStream → PutRef
+//! ```
+//!
+//! First argument after subcommand flags is passed through to cargo verbatim,
+//! so `claw-cargo build --profile release -p my-crate` works the same as
+//! `cargo build --profile release -p my-crate` — the only difference is
+//! the pre/post cache lookup.
+
+mod cluster;
+mod cargo_init;
+mod config;
+mod head_watch;
+mod hot;
+mod manifest;
+mod restore;
+mod snapshot;
+mod sync;
+mod zfs;
+mod actions;
+mod daemon;
+mod serve;
+
+use anyhow::{bail, Context, Result};
+use clap::{Parser, Subcommand};
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::Command;
+use std::time::Instant;
+
+use crate::cluster::blob::BlobId;
+use crate::cluster::build_cache::{
+    capture_target, compute_workspace_fingerprint, restore_target, Fingerprint,
+};
+use crate::cluster::rpc::{
+    call_blob_get_stream, call_blob_put_stream, call_blob_stat, call_get_ref, call_put_ref,
+};
+use crate::cluster::transport::{NodeIdentity, QuicClient};
+
+#[derive(Parser)]
+#[command(
+    name = "claw-cargo",
+    about = "Fingerprint-keyed cargo build cache — wraps `cargo build` with a peer lookup",
+    version
+)]
+struct Cli {
+    #[command(subcommand)]
+    cmd: Cmd,
+}
+
+#[derive(Subcommand)]
+enum Cmd {
+    /// Build with cache lookup: hit → restore + cargo build (fast);
+    /// miss → cargo build + capture + upload.
+    Build(BuildArgs),
+    /// Print the fingerprint for the current workspace without touching
+    /// the cache. Useful for diagnosis + integration tests.
+    Fingerprint {
+        #[arg(long, default_value = "dev")]
+        profile: String,
+        #[arg(long, num_args = 0.., value_delimiter = ',')]
+        features: Vec<String>,
+        #[arg(long)]
+        workspace: Option<PathBuf>,
+    },
+}
+
+#[derive(clap::Args)]
+struct BuildArgs {
+    /// Peer's RPC socket (gossip_port + 1 by default).
+    #[arg(long)]
+    peer_addr: SocketAddr,
+    /// Peer's node name — must match the peer's leaf cert SAN.
+    #[arg(long)]
+    peer: String,
+    /// Directory holding this node's mTLS material
+    /// (`ca.crt` + `node.crt` + `node.key`).
+    #[arg(long)]
+    tls_dir: PathBuf,
+    /// Cargo profile — passed to cargo AND used in the fingerprint.
+    #[arg(long, default_value = "dev")]
+    profile: String,
+    /// Enabled features — passed to cargo AND used in the fingerprint.
+    #[arg(long, num_args = 0.., value_delimiter = ',')]
+    features: Vec<String>,
+    /// Workspace root. Defaults to the current directory.
+    #[arg(long)]
+    workspace: Option<PathBuf>,
+    /// Skip capture + upload on a miss. Useful for read-only cache use.
+    #[arg(long)]
+    no_upload: bool,
+    /// Extra args passed verbatim to `cargo build`.
+    #[arg(last = true)]
+    cargo_args: Vec<String>,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into()))
+        .init();
+    let cli = Cli::parse();
+    match cli.cmd {
+        Cmd::Build(args) => cmd_build(args).await,
+        Cmd::Fingerprint {
+            profile,
+            features,
+            workspace,
+        } => cmd_fingerprint(profile, features, workspace),
+    }
+}
+
+fn cmd_fingerprint(
+    profile: String,
+    features: Vec<String>,
+    workspace: Option<PathBuf>,
+) -> Result<()> {
+    let workspace =
+        workspace.unwrap_or_else(|| std::env::current_dir().expect("cwd"));
+    let (_inputs, fp) =
+        compute_workspace_fingerprint(&workspace, &profile, &features)?;
+    println!("workspace:   {}", workspace.display());
+    println!("profile:     {}", profile);
+    println!("features:    {}", features.join(","));
+    println!("fingerprint: {}", fp);
+    Ok(())
+}
+
+async fn cmd_build(args: BuildArgs) -> Result<()> {
+    let workspace = args
+        .workspace
+        .clone()
+        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
+    let target_dir = workspace.join("target").join(&args.profile);
+
+    // ── 1. Fingerprint ────────────────────────────────────────────
+    let (_inputs, fingerprint) =
+        compute_workspace_fingerprint(&workspace, &args.profile, &args.features)
+            .context("computing workspace fingerprint")?;
+    tracing::info!("fingerprint {}", fingerprint);
+
+    // ── 2. Connect to peer ────────────────────────────────────────
+    let identity = NodeIdentity::from_pem_dir(&args.tls_dir).with_context(|| {
+        format!("loading node identity from {}", args.tls_dir.display())
+    })?;
+    let client = QuicClient::new("0.0.0.0:0".parse()?, identity)?;
+    let conn = client.connect(args.peer_addr, &args.peer).await?;
+
+    // ── 3. Cache lookup ───────────────────────────────────────────
+    let key = *fingerprint.as_bytes();
+    let mut outcome = CacheOutcome::Miss;
+    match call_get_ref(&conn, &key).await? {
+        Some(value_bytes) => {
+            let blob_id = BlobId::from_bytes(value_bytes);
+            tracing::info!("cache HIT — blob {}", blob_id);
+            match call_blob_stat(&conn, &blob_id).await? {
+                Some(stat) => {
+                    tracing::info!(
+                        "downloading cached target ({} chunks, {} bytes) → {}",
+                        stat.chunk_count,
+                        stat.total_size,
+                        target_dir.display()
+                    );
+                    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::warn!(
+                            "ref pointed at blob {} but peer returned NotFound; falling through",
+                            blob_id
+                        );
+                    } else {
+                        std::fs::create_dir_all(&target_dir).with_context(|| {
+                            format!("creating {}", target_dir.display())
+                        })?;
+                        restore_target(&buf, &target_dir)
+                            .context("restoring cached target dir")?;
+                        outcome = CacheOutcome::Hit {
+                            blob_id,
+                            downloaded_bytes: buf.len() as u64,
+                        };
+                    }
+                }
+                None => {
+                    tracing::warn!(
+                        "ref pointed at blob {} but peer has no such blob; falling through",
+                        blob_id
+                    );
+                }
+            }
+        }
+        None => {
+            tracing::info!("cache MISS — no ref for fingerprint");
+        }
+    }
+
+    // ── 4. Run cargo ──────────────────────────────────────────────
+    let cargo_started = Instant::now();
+    let cargo_status = run_cargo(&workspace, &args.profile, &args.features, &args.cargo_args)?;
+    let cargo_elapsed = cargo_started.elapsed();
+    if !cargo_status.success() {
+        bail!("cargo build failed with exit {}", cargo_status);
+    }
+    tracing::info!("cargo build finished in {:?}", cargo_elapsed);
+
+    // ── 5. On miss, capture + upload ──────────────────────────────
+    if matches!(outcome, CacheOutcome::Miss) && !args.no_upload {
+        if !target_dir.is_dir() {
+            tracing::warn!(
+                "target dir {} does not exist after cargo build — skipping upload",
+                target_dir.display()
+            );
+        } else {
+            let capture_started = Instant::now();
+            let bytes = capture_target(&target_dir).context("capturing target dir")?;
+            let capture_elapsed = capture_started.elapsed();
+            tracing::info!(
+                "captured target: {} bytes in {:?}",
+                bytes.len(),
+                capture_elapsed
+            );
+
+            let upload_started = Instant::now();
+            let cursor = std::io::Cursor::new(bytes.clone());
+            let blob_id = call_blob_put_stream(&conn, cursor).await?;
+            let upload_elapsed = upload_started.elapsed();
+            tracing::info!(
+                "uploaded blob {} in {:?}",
+                blob_id,
+                upload_elapsed
+            );
+
+            call_put_ref(&conn, &key, blob_id.as_bytes()).await?;
+            tracing::info!("set ref {} → {}", fingerprint, blob_id);
+
+            outcome = CacheOutcome::Populated {
+                blob_id,
+                uploaded_bytes: bytes.len() as u64,
+            };
+        }
+    }
+
+    // ── 6. Shutdown + summary ─────────────────────────────────────
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+
+    print_summary(&fingerprint, &outcome, cargo_elapsed);
+    Ok(())
+}
+
+/// Outcome of the cache lookup + build cycle. Reported at end-of-run.
+enum CacheOutcome {
+    Hit {
+        blob_id: BlobId,
+        downloaded_bytes: u64,
+    },
+    Miss,
+    Populated {
+        blob_id: BlobId,
+        uploaded_bytes: u64,
+    },
+}
+
+fn print_summary(fp: &Fingerprint, outcome: &CacheOutcome, cargo_elapsed: std::time::Duration) {
+    println!();
+    println!("── claw-cargo summary ──────────────────────────────");
+    println!("fingerprint:  {}", fp);
+    match outcome {
+        CacheOutcome::Hit {
+            blob_id,
+            downloaded_bytes,
+        } => {
+            println!("cache:        HIT ({} bytes downloaded)", downloaded_bytes);
+            println!("blob:         {}", blob_id);
+        }
+        CacheOutcome::Miss => {
+            println!("cache:        MISS (no upload performed)");
+        }
+        CacheOutcome::Populated {
+            blob_id,
+            uploaded_bytes,
+        } => {
+            println!(
+                "cache:        MISS → populated ({} bytes uploaded)",
+                uploaded_bytes
+            );
+            println!("blob:         {}", blob_id);
+        }
+    }
+    println!("cargo build:  {:?}", cargo_elapsed);
+    println!("────────────────────────────────────────────────────");
+}
+
+/// Invoke `cargo build` with the same profile/features + any passthrough
+/// args the caller supplied. Runs to completion, returns the exit status.
+fn run_cargo(
+    workspace: &std::path::Path,
+    profile: &str,
+    features: &[String],
+    passthrough: &[String],
+) -> Result<std::process::ExitStatus> {
+    let mut cmd = Command::new("cargo");
+    cmd.current_dir(workspace).arg("build").arg("--profile").arg(profile);
+    if !features.is_empty() {
+        cmd.arg("--features").arg(features.join(","));
+    }
+    for arg in passthrough {
+        cmd.arg(arg);
+    }
+    let status = cmd.status().context("spawning cargo build")?;
+    Ok(status)
+}

claw-store/src/cluster.rs

@@ -17,6 +17,7 @@
 pub mod blob;
 pub mod build_cache;
 pub mod gossip;
+pub mod refs;
 pub mod rpc;
 pub mod services;
 pub mod transport;

claw-store/src/cluster/refs.rs

@@ -0,0 +1,243 @@
+//! Reference store — 32-byte key → 32-byte value mapping (Phase 5b).
+//!
+//! Used to map a [`Fingerprint`](crate::cluster::build_cache::Fingerprint)
+//! to the [`BlobId`](crate::cluster::blob::BlobId) of its cached
+//! artifact. Deliberately a dumb primitive: no versioning, no CRDT
+//! semantics — Phase 3 will layer a richer metadata model on top,
+//! but every real cargo-cache lookup we need in Phase 5 is a single
+//! key → single value.
+//!
+//! # Layout
+//!
+//! ```text
+//! <root>/
+//!   refs/<kk>/<key_hex>.ref     — 32 raw bytes (the value)
+//!   .tmp/                        — atomic-rename staging
+//! ```
+//!
+//! `<kk>` is the first two hex chars of the key, keeping directory
+//! fan-out bounded (256 entries per level). Writes go through
+//! tempfile + rename so a mid-write crash leaves either a complete
+//! file or nothing.
+
+use anyhow::{bail, Context, Result};
+use std::path::{Path, PathBuf};
+use tokio::io::AsyncWriteExt;
+
+/// A raw 32-byte key. Callers wrap semantically — the store itself
+/// treats keys as opaque bytes.
+pub type RefKey = [u8; 32];
+
+/// A raw 32-byte value.
+pub type RefValue = [u8; 32];
+
+/// Directory-backed reference store.
+#[derive(Debug, Clone)]
+pub struct RefStore {
+    root: PathBuf,
+}
+
+impl RefStore {
+    /// Open (create if missing) a ref store rooted at `root`. Creates
+    /// `refs/` and `.tmp/` subdirs. Safe to call on an existing store.
+    pub fn open(root: PathBuf) -> Result<Self> {
+        std::fs::create_dir_all(root.join("refs"))
+            .with_context(|| format!("creating refs dir under {}", root.display()))?;
+        std::fs::create_dir_all(root.join(".tmp"))
+            .with_context(|| format!("creating .tmp dir under {}", root.display()))?;
+        Ok(Self { root })
+    }
+
+    pub fn root(&self) -> &Path {
+        &self.root
+    }
+
+    /// Look up a key. `None` when no ref has been set. Errors only on
+    /// filesystem failures other than "not found".
+    pub async fn get(&self, key: &RefKey) -> Result<Option<RefValue>> {
+        let path = self.ref_path(key);
+        match tokio::fs::read(&path).await {
+            Ok(bytes) => {
+                if bytes.len() != 32 {
+                    bail!(
+                        "ref at {} has wrong length {} (expected 32)",
+                        path.display(),
+                        bytes.len()
+                    );
+                }
+                let mut out = [0u8; 32];
+                out.copy_from_slice(&bytes);
+                Ok(Some(out))
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(e) => Err(anyhow::Error::from(e))
+                .with_context(|| format!("reading ref at {}", path.display())),
+        }
+    }
+
+    /// Set a key → value mapping. Overwrites any existing value —
+    /// callers who need last-writer-wins protection should implement
+    /// it in the layer above (Phase 3 CRDT). Atomic on the filesystem:
+    /// a mid-write crash leaves the previous value intact.
+    pub async fn put(&self, key: &RefKey, value: &RefValue) -> Result<()> {
+        let final_path = self.ref_path(key);
+        if let Some(parent) = final_path.parent() {
+            tokio::fs::create_dir_all(parent).await.with_context(|| {
+                format!("creating ref bucket {}", parent.display())
+            })?;
+        }
+        self.atomic_write(&final_path, value).await
+    }
+
+    /// Delete a key. Returns `true` if a ref was removed, `false` if
+    /// no ref existed for the key.
+    pub async fn delete(&self, key: &RefKey) -> Result<bool> {
+        let path = self.ref_path(key);
+        match tokio::fs::remove_file(&path).await {
+            Ok(()) => Ok(true),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+            Err(e) => Err(anyhow::Error::from(e)),
+        }
+    }
+
+    /// Whether a value is set for the given key.
+    pub async fn contains(&self, key: &RefKey) -> Result<bool> {
+        match tokio::fs::metadata(self.ref_path(key)).await {
+            Ok(_) => Ok(true),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+            Err(e) => Err(anyhow::Error::from(e)),
+        }
+    }
+
+    fn ref_path(&self, key: &RefKey) -> PathBuf {
+        let hex = hex32(key);
+        self.root
+            .join("refs")
+            .join(&hex[..2])
+            .join(format!("{hex}.ref"))
+    }
+
+    async fn atomic_write(&self, final_path: &Path, bytes: &[u8]) -> Result<()> {
+        let tmp_dir = self.root.join(".tmp");
+        let tmp_name = format!(
+            "{}.{}",
+            std::process::id(),
+            RANDOM_SUFFIX.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+        );
+        let tmp_path = tmp_dir.join(tmp_name);
+        {
+            let mut f = tokio::fs::File::create(&tmp_path).await.with_context(|| {
+                format!("creating tmp ref {}", tmp_path.display())
+            })?;
+            f.write_all(bytes).await?;
+            f.sync_all().await?;
+        }
+        tokio::fs::rename(&tmp_path, final_path)
+            .await
+            .with_context(|| {
+                format!(
+                    "renaming {} → {}",
+                    tmp_path.display(),
+                    final_path.display()
+                )
+            })?;
+        Ok(())
+    }
+}
+
+static RANDOM_SUFFIX: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
+fn hex32(bytes: &[u8; 32]) -> String {
+    let mut out = String::with_capacity(64);
+    for b in bytes {
+        out.push_str(&format!("{b:02x}"));
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn open() -> (tempfile::TempDir, RefStore) {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let store = RefStore::open(tmp.path().to_path_buf()).unwrap();
+        (tmp, store)
+    }
+
+    #[tokio::test]
+    async fn open_creates_layout() {
+        let (tmp, store) = open();
+        assert!(tmp.path().join("refs").is_dir());
+        assert!(tmp.path().join(".tmp").is_dir());
+        assert_eq!(store.root(), tmp.path());
+    }
+
+    #[tokio::test]
+    async fn get_returns_none_for_missing() {
+        let (_tmp, store) = open();
+        let key = [0u8; 32];
+        assert_eq!(store.get(&key).await.unwrap(), None);
+        assert!(!store.contains(&key).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn put_and_get_round_trip() {
+        let (_tmp, store) = open();
+        let key = [0x11u8; 32];
+        let value = [0x22u8; 32];
+        store.put(&key, &value).await.unwrap();
+        assert_eq!(store.get(&key).await.unwrap(), Some(value));
+        assert!(store.contains(&key).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn put_overwrites_prior_value() {
+        let (_tmp, store) = open();
+        let key = [0xaau8; 32];
+        let v1 = [0x01u8; 32];
+        let v2 = [0x02u8; 32];
+        store.put(&key, &v1).await.unwrap();
+        store.put(&key, &v2).await.unwrap();
+        assert_eq!(store.get(&key).await.unwrap(), Some(v2));
+    }
+
+    #[tokio::test]
+    async fn delete_removes_ref_and_reports() {
+        let (_tmp, store) = open();
+        let key = [0x55u8; 32];
+        let value = [0x66u8; 32];
+        store.put(&key, &value).await.unwrap();
+        assert!(store.delete(&key).await.unwrap());
+        assert_eq!(store.get(&key).await.unwrap(), None);
+        // Second delete: returns false, not an error.
+        assert!(!store.delete(&key).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn distinct_keys_produce_distinct_files() {
+        // Verify the on-disk layout by inspection — different keys
+        // land in different bucket dirs (proves the fan-out heuristic).
+        let (tmp, store) = open();
+        let k1 = [0xffu8; 32];
+        let k2 = [0x00u8; 32];
+        store.put(&k1, &[0u8; 32]).await.unwrap();
+        store.put(&k2, &[0u8; 32]).await.unwrap();
+        assert!(tmp.path().join("refs/ff").exists());
+        assert!(tmp.path().join("refs/00").exists());
+    }
+
+    #[tokio::test]
+    async fn rejects_wrong_length_on_disk() {
+        // Simulate corruption / hand-tampering: a file at the ref
+        // path exists but isn't 32 bytes. Read must surface an error,
+        // not silently return garbage.
+        let (tmp, store) = open();
+        let key = [0x7fu8; 32];
+        let path = store.ref_path(&key);
+        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+        std::fs::write(&path, b"only 4b").unwrap();
+        let err = store.get(&key).await.unwrap_err().to_string();
+        assert!(err.contains("wrong length"), "unexpected: {err}");
+    }
+}

claw-store/src/cluster/rpc.rs

@@ -27,6 +27,7 @@
 
 use crate::cluster::blob::{BlobId, BlobManifest, BlobStat, BlobStore, ChunkHash};
 use crate::cluster::gossip::{ClusterGossip, PeerView};
+use crate::cluster::refs::{RefKey, RefStore, RefValue};
 use anyhow::{bail, Context, Result};
 use quinn::{Connection, ConnectionError};
 use serde::{Deserialize, Serialize};
@@ -97,6 +98,18 @@ pub enum Method {
     /// `missing` list means the manifest was written; a non-empty
     /// list tells the client which chunks to upload before retrying.
     PutManifest = 0x0c,
+    /// Phase 5b: fetch a 32-byte value keyed by a 32-byte reference key.
+    /// Used by the fingerprint-keyed cargo cache: key = fingerprint,
+    /// value = BlobId of the cached target-dir tarball.
+    /// `payload`: 32-byte `RefKey`.
+    /// Reply: 32 bytes on hit, single-byte [`ErrorCode::NotFound`] on miss.
+    GetRef = 0x0d,
+    /// Phase 5b: set the value for a reference key. Overwrites any
+    /// prior value — the CRDT/versioning semantics come later
+    /// (Phase 3). `payload`: 32-byte `RefKey` || 32-byte `RefValue`.
+    /// Reply: `STREAM_STATUS_OK` (1 byte) on success, or a
+    /// single-byte error code.
+    PutRef = 0x0e,
 }
 
 impl Method {
@@ -116,6 +129,8 @@ impl Method {
             0x0a => Some(Method::PutChunk),
             0x0b => Some(Method::GetChunk),
             0x0c => Some(Method::PutManifest),
+            0x0d => Some(Method::GetRef),
+            0x0e => Some(Method::PutRef),
             _ => None,
         }
     }
@@ -201,6 +216,7 @@ pub struct PutManifestReply {
 pub struct RpcRouter {
     gossip: Arc<ClusterGossip>,
     blob_store: Option<Arc<BlobStore>>,
+    ref_store: Option<Arc<RefStore>>,
     local_name: String,
     local_zone: String,
 }
@@ -218,6 +234,7 @@ impl RpcRouter {
         Self {
             gossip,
             blob_store: None,
+            ref_store: None,
             local_name,
             local_zone,
         }
@@ -230,10 +247,21 @@ impl RpcRouter {
         self
     }
 
+    /// Attach a local ref store. Enables the `GetRef` / `PutRef`
+    /// methods.
+    pub fn with_ref_store(mut self, store: Arc<RefStore>) -> Self {
+        self.ref_store = Some(store);
+        self
+    }
+
     pub fn blob_store(&self) -> Option<&Arc<BlobStore>> {
         self.blob_store.as_ref()
     }
 
+    pub fn ref_store(&self) -> Option<&Arc<RefStore>> {
+        self.ref_store.as_ref()
+    }
+
     /// Dispatch a single request. Called by [`serve_connection`] for
     /// every accepted bidi stream. Test code may call it directly to
     /// bypass the transport.
@@ -426,8 +454,48 @@ impl RpcRouter {
                     .context("encoding PutManifestReply as JSON")?;
                 Ok(HandlerOutcome::Reply(json))
             }
+            Method::GetRef => {
+                let store = match &self.ref_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let key = match decode_32(payload) {
+                    Some(k) => k,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                match store.get(&key).await? {
+                    Some(value) => Ok(HandlerOutcome::Reply(value.to_vec())),
+                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
                 }
             }
+            Method::PutRef => {
+                let store = match &self.ref_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                if payload.len() != 64 {
+                    return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest));
+                }
+                let mut key = [0u8; 32];
+                let mut value = [0u8; 32];
+                key.copy_from_slice(&payload[..32]);
+                value.copy_from_slice(&payload[32..]);
+                store.put(&key, &value).await?;
+                Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
+            }
+        }
+    }
+}
+
+/// Parse a payload as a 32-byte array. Shared by `GetRef` and any
+/// future single-32-byte-payload methods.
+fn decode_32(payload: &[u8]) -> Option<[u8; 32]> {
+    if payload.len() != 32 {
+        return None;
+    }
+    let mut out = [0u8; 32];
+    out.copy_from_slice(payload);
+    Some(out)
 }
 
 /// Parse a payload as a 32-byte ChunkHash. Same shape as
@@ -999,6 +1067,54 @@ pub async fn call_put_manifest(
     Ok(decoded.missing)
 }
 
+// ── Phase 5b: reference-store client helpers ─────────────────────────
+
+/// Look up a 32-byte value by 32-byte key. `Ok(None)` on `NotFound`.
+pub async fn call_get_ref(
+    conn: &Connection,
+    key: &RefKey,
+) -> Result<Option<RefValue>> {
+    let reply = rpc_call(conn, Method::GetRef, key).await?;
+    if reply.len() == 1 {
+        match decode_error(reply[0]) {
+            Some(ErrorCode::NotFound) => return Ok(None),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => {}
+        }
+    }
+    if reply.len() != 32 {
+        bail!("expected 32-byte ref value, got {} bytes", reply.len());
+    }
+    let mut value = [0u8; 32];
+    value.copy_from_slice(&reply);
+    Ok(Some(value))
+}
+
+/// Set a 32-byte value for a 32-byte key. Overwrites any prior value.
+pub async fn call_put_ref(
+    conn: &Connection,
+    key: &RefKey,
+    value: &RefValue,
+) -> Result<()> {
+    let mut payload = Vec::with_capacity(64);
+    payload.extend_from_slice(key);
+    payload.extend_from_slice(value);
+    let reply = rpc_call(conn, Method::PutRef, &payload).await?;
+    if reply.len() != 1 {
+        bail!("expected single-byte PutRef reply, got {} bytes", reply.len());
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(()),
+        code => match decode_error(code) {
+            Some(err) => bail!("peer rejected PutRef: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for PutRef",
+                code
+            ),
+        },
+    }
+}
+
 /// High-level partial-sync helper: replicate a local blob to a peer
 /// by uploading only chunks the peer is missing.
 ///

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

@@ -725,6 +725,139 @@ async fn end_to_end_push_blob_missing_chunks_replicates_only_needed_bytes() {
     accept_task.abort();
 }
 
+// ── Phase 5b: reference-store RPC ────────────────────────────────────
+
+async fn router_with_blobs_and_refs(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
+    use crate::cluster::refs::RefStore;
+    let gossip = bootstrap_gossip(name, port).await;
+    let tmp = tempfile::TempDir::new().unwrap();
+    let blob_store =
+        Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
+    let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
+    let router = Arc::new(
+        RpcRouter::new(gossip, name.into(), "fabric-10g".into())
+            .with_blob_store(blob_store)
+            .with_ref_store(ref_store),
+    );
+    (tmp, router)
+}
+
+#[test]
+fn phase_5b_method_byte_encoding() {
+    assert_eq!(Method::GetRef.as_byte(), 0x0d);
+    assert_eq!(Method::PutRef.as_byte(), 0x0e);
+    assert_eq!(Method::from_byte(0x0d), Some(Method::GetRef));
+    assert_eq!(Method::from_byte(0x0e), Some(Method::PutRef));
+}
+
+#[tokio::test]
+async fn get_ref_returns_not_found_for_missing() {
+    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
+    let key = [0u8; 32];
+    let mut req = vec![Method::GetRef.as_byte()];
+    req.extend_from_slice(&key);
+    assert_eq!(
+        dispatch(&router, &req).await,
+        vec![ErrorCode::NotFound.as_byte()]
+    );
+}
+
+#[tokio::test]
+async fn put_ref_stores_and_get_ref_reads_back() {
+    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
+    let key = [0x11u8; 32];
+    let value = [0x22u8; 32];
+    let mut put = vec![Method::PutRef.as_byte()];
+    put.extend_from_slice(&key);
+    put.extend_from_slice(&value);
+    assert_eq!(dispatch(&router, &put).await, vec![STREAM_STATUS_OK]);
+
+    let mut get = vec![Method::GetRef.as_byte()];
+    get.extend_from_slice(&key);
+    assert_eq!(dispatch(&router, &get).await, value.to_vec());
+}
+
+#[tokio::test]
+async fn put_ref_rejects_wrong_length_payload() {
+    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
+    // 63 bytes — one shy of the 32+32 requirement.
+    let req = {
+        let mut r = vec![Method::PutRef.as_byte()];
+        r.extend(vec![0u8; 63]);
+        r
+    };
+    assert_eq!(
+        dispatch(&router, &req).await,
+        vec![ErrorCode::InvalidRequest.as_byte()]
+    );
+}
+
+#[tokio::test]
+async fn get_ref_rejects_wrong_length_payload() {
+    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
+    let req = vec![Method::GetRef.as_byte(), 0, 1, 2];
+    assert_eq!(
+        dispatch(&router, &req).await,
+        vec![ErrorCode::InvalidRequest.as_byte()]
+    );
+}
+
+#[tokio::test]
+async fn ref_rpcs_return_not_configured_without_store() {
+    let gossip = bootstrap_gossip("solo", next_port()).await;
+    // Router with a blob store but NO ref store.
+    let tmp = tempfile::TempDir::new().unwrap();
+    let blob =
+        Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().to_path_buf()).unwrap());
+    let router = Arc::new(
+        RpcRouter::new(gossip, "solo".into(), "z".into()).with_blob_store(blob),
+    );
+
+    for method in [Method::GetRef, Method::PutRef] {
+        let mut req = vec![method.as_byte()];
+        req.extend_from_slice(&[0u8; 32]);
+        req.extend_from_slice(&[0u8; 32]);
+        assert_eq!(
+            dispatch(&router, &req).await,
+            vec![ErrorCode::NotConfigured.as_byte()],
+            "method {method:?} should be NotConfigured"
+        );
+    }
+}
+
+#[tokio::test]
+async fn end_to_end_put_ref_get_ref_over_real_quic() {
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp, router) = router_with_blobs_and_refs("a", next_port()).await;
+
+    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();
+
+    let key = [0x77u8; 32];
+    let value = [0x88u8; 32];
+
+    // Miss first.
+    assert!(call_get_ref(&conn, &key).await.unwrap().is_none());
+    // Put.
+    call_put_ref(&conn, &key, &value).await.unwrap();
+    // Hit.
+    assert_eq!(call_get_ref(&conn, &key).await.unwrap(), Some(value));
+
+    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 call_get_chunk_verifies_returned_hash() {
     let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();

claw-store/src/cluster/services.rs

@@ -14,6 +14,7 @@
 
 use crate::cluster::blob::BlobStore;
 use crate::cluster::gossip::ClusterGossip;
+use crate::cluster::refs::RefStore;
 use crate::cluster::rpc::{serve_connection, RpcRouter};
 use crate::cluster::transport::{NodeIdentity, QuicServer};
 use crate::config::ClusterConfig;
@@ -44,6 +45,11 @@ pub struct ClusterServices {
     /// time. `None` means this node runs gossip-only (Blob RPCs return
     /// [`crate::cluster::rpc::ErrorCode::NotConfigured`]).
     pub blob_store: Option<Arc<BlobStore>>,
+    /// Local ref store (Phase 5b). Backs the `GetRef`/`PutRef` RPCs
+    /// used by the fingerprint-keyed cargo cache. Opened automatically
+    /// alongside the blob store — a node with one gets the other, so
+    /// the entire Phase 5 substrate is enabled by a single config field.
+    pub ref_store: Option<Arc<RefStore>>,
     /// QUIC accept-loop task. `None` when `[cluster.tls]` was absent
     /// and RPC therefore didn't come up.
     accept_task: Option<JoinHandle<()>>,
@@ -90,7 +96,7 @@ impl ClusterServices {
         // outside the TLS branch: a node can serve blobs to callers
         // without RPC (via in-process API) or over RPC (once TLS is
         // configured too).
-        let blob_store: Option<Arc<BlobStore>> = match blob_store_root {
+        let blob_store: Option<Arc<BlobStore>> = match blob_store_root.as_ref() {
             Some(root) => {
                 let store = BlobStore::open(root.clone())
                     .with_context(|| format!("opening blob store at {}", root.display()))?;
@@ -103,6 +109,21 @@ impl ClusterServices {
             }
         };
 
+        // Ref store lives under `<blob_root>/refs-db` so the entire
+        // Phase 5 substrate is enabled by a single config field.
+        // Nodes without a blob store don't get a ref store either;
+        // the fingerprint cache is meaningless without content.
+        let ref_store: Option<Arc<RefStore>> = match blob_store_root.as_ref() {
+            Some(root) => {
+                let refs_dir = root.join("refs-db");
+                let store = RefStore::open(refs_dir.clone())
+                    .with_context(|| format!("opening ref store at {}", refs_dir.display()))?;
+                tracing::info!("ref store opened at {}", refs_dir.display());
+                Some(Arc::new(store))
+            }
+            None => None,
+        };
+
         // RPC server + accept loop — only when TLS material is configured.
         let accept_task = match &cluster.tls {
             Some(tls) => {
@@ -123,6 +144,9 @@ impl ClusterServices {
                 if let Some(store) = &blob_store {
                     router = router.with_blob_store(store.clone());
                 }
+                if let Some(store) = &ref_store {
+                    router = router.with_ref_store(store.clone());
+                }
                 let router = Arc::new(router);
                 tracing::info!("cluster RPC server listening on {}", bind);
                 Some(tokio::spawn(async move {
@@ -152,6 +176,7 @@ impl ClusterServices {
         Ok(Self {
             gossip,
             blob_store,
+            ref_store,
             accept_task,
             metric_task,
         })
@@ -162,6 +187,11 @@ impl ClusterServices {
         self.blob_store.is_some()
     }
 
+    /// Whether a local ref store was configured at start time.
+    pub fn ref_store_enabled(&self) -> bool {
+        self.ref_store.is_some()
+    }
+
     /// Whether the QUIC RPC server is running. `false` when `[cluster.tls]`
     /// was absent at start time.
     pub fn rpc_enabled(&self) -> bool {