Merge pull request 'Phase 2: content-addressed blob store' (#6) from phase-2-blob-store into main

Reviewed-on: #6

Cargo.lock

@@ -82,6 +82,18 @@ version = "1.0.102"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
 
+[[package]]
+name = "arrayref"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76a2e8124351fda1ef8aaaa3bbd7ebbcb486bbcd4225aca0aa0d84bb2db8fecb"
+
+[[package]]
+name = "arrayvec"
+version = "0.7.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
+
 [[package]]
 name = "asn1-rs"
 version = "0.6.2"
@@ -223,6 +235,20 @@ version = "2.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8"
 
+[[package]]
+name = "blake3"
+version = "1.8.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0aa83c34e62843d924f905e0f5c866eb1dd6545fc4d719e803d9ba6030371fce"
+dependencies = [
+ "arrayref",
+ "arrayvec",
+ "cc",
+ "cfg-if",
+ "constant_time_eq",
+ "cpufeatures",
+]
+
 [[package]]
 name = "bumpalo"
 version = "3.20.3"
@@ -349,6 +375,7 @@ version = "0.3.0"
 dependencies = [
  "anyhow",
  "axum",
+ "blake3",
  "chitchat",
  "chrono",
  "clap",
@@ -377,6 +404,12 @@ version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 
+[[package]]
+name = "constant_time_eq"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b"
+
 [[package]]
 name = "core-foundation-sys"
 version = "0.8.7"

claw-store/Cargo.toml

@@ -47,6 +47,11 @@ rcgen = { version = "0.13", features = ["pem", "x509-parser"] }
 # NodeIdentity::from_pem_files (Phase 1d) so persistent identity round-trips
 # through the file system on daemon restart.
 rustls-pemfile = "2"
+# v1 — content-addressed hashing for the Phase 2 blob store. BLAKE3 is
+# used both for whole-blob addressing (BlobId) and for per-chunk
+# addressing (dedup). Fast enough that a full-blob rehash on read
+# verification stays cheap.
+blake3 = "1"
 
 [dev-dependencies]
 tempfile = "3"

claw-store/src/cluster.rs

@@ -14,6 +14,7 @@
 //! after ~5 min so route changes (node moved networks, LAN NIC came back)
 //! propagate without a daemon restart.
 
+pub mod blob;
 pub mod gossip;
 pub mod rpc;
 pub mod services;

claw-store/src/cluster/blob.rs

@@ -0,0 +1,802 @@
+//! Content-addressed blob store (Phase 2).
+//!
+//! Every blob is identified by the BLAKE3 hash of its full content:
+//! same bytes → same `BlobId`, always. Blobs are physically stored as
+//! a sequence of BLAKE3-hashed 4 MB chunks, so two blobs that share a
+//! prefix (say two cargo target dirs with 95% of the same deps) share
+//! storage at chunk granularity without any special detection logic.
+//!
+//! # Layout
+//!
+//! ```text
+//! <root>/
+//!   blobs/
+//!     <bb>/<blob_id>.manifest.json      { chunks: [<chunk_hash>...], total_size: u64 }
+//!   chunks/
+//!     <cc>/<chunk_hash>                  raw bytes of one chunk
+//!   .tmp/                               atomic-rename staging area
+//! ```
+//!
+//! `<bb>` and `<cc>` are the first two hex chars of the corresponding
+//! hash. Keeps directory fan-out bounded (256 entries per level) —
+//! important on a warm-tier ZFS dataset with tens of thousands of blobs.
+//!
+//! Writes use a temp file + rename so a crash mid-write leaves either
+//! a complete file or nothing (never a truncated partial). Chunks are
+//! immutable: if `<chunk_hash>` already exists, we skip re-writing it.
+//!
+//! # Deletion + GC
+//!
+//! [`BlobStore::delete_manifest`] removes the top-level manifest but
+//! leaves chunks — orphan cleanup is a separate sweep so we avoid
+//! walking the whole store for every deletion. Run
+//! [`BlobStore::gc_orphan_chunks`] periodically (nightly is fine) to
+//! reclaim any chunks no longer referenced by a live manifest.
+
+use anyhow::{bail, Context, Result};
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+use tokio::io::AsyncWriteExt;
+
+/// Physical chunk size. 4 MB is the sweet spot for our workloads:
+/// small enough that dedup catches typical file-tree overlaps between
+/// similar cargo build outputs, large enough that per-chunk overhead
+/// (hash compute + syscall) stays negligible on hot-path reads.
+pub const CHUNK_SIZE: usize = 4 * 1024 * 1024;
+
+/// Content-addressed identifier for a blob: the BLAKE3 hash of its full
+/// content. Compares/orderings are on the raw 32-byte hash, so
+/// serialisation to hex is only for display and on-disk paths.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct BlobId([u8; 32]);
+
+impl BlobId {
+    pub fn from_bytes(hash: [u8; 32]) -> Self {
+        Self(hash)
+    }
+
+    pub fn as_bytes(&self) -> &[u8; 32] {
+        &self.0
+    }
+
+    /// Lowercase hex string, 64 chars long — the on-disk path suffix.
+    pub fn to_hex(&self) -> String {
+        hex_encode(&self.0)
+    }
+
+    /// Parse from lowercase hex. Rejects wrong-length or non-hex input.
+    pub fn from_hex(s: &str) -> Result<Self> {
+        let bytes = hex_decode(s)?;
+        if bytes.len() != 32 {
+            bail!(
+                "BlobId hex must be exactly 64 chars (32 bytes); got {}",
+                bytes.len() * 2
+            );
+        }
+        let mut arr = [0u8; 32];
+        arr.copy_from_slice(&bytes);
+        Ok(Self(arr))
+    }
+}
+
+impl std::fmt::Display for BlobId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.to_hex())
+    }
+}
+
+impl Serialize for BlobId {
+    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
+        s.serialize_str(&self.to_hex())
+    }
+}
+
+impl<'de> Deserialize<'de> for BlobId {
+    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
+        let s = String::deserialize(d)?;
+        BlobId::from_hex(&s).map_err(serde::de::Error::custom)
+    }
+}
+
+/// Physical hash of a single chunk. Same shape as [`BlobId`] but a
+/// distinct type so we can't accidentally lookup a chunk with a blob
+/// hash or vice versa.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct ChunkHash([u8; 32]);
+
+impl ChunkHash {
+    pub fn from_bytes(hash: [u8; 32]) -> Self {
+        Self(hash)
+    }
+
+    pub fn as_bytes(&self) -> &[u8; 32] {
+        &self.0
+    }
+
+    pub fn to_hex(&self) -> String {
+        hex_encode(&self.0)
+    }
+
+    pub fn from_hex(s: &str) -> Result<Self> {
+        let bytes = hex_decode(s)?;
+        if bytes.len() != 32 {
+            bail!(
+                "ChunkHash hex must be exactly 64 chars (32 bytes); got {}",
+                bytes.len() * 2
+            );
+        }
+        let mut arr = [0u8; 32];
+        arr.copy_from_slice(&bytes);
+        Ok(Self(arr))
+    }
+}
+
+impl Serialize for ChunkHash {
+    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
+        s.serialize_str(&self.to_hex())
+    }
+}
+
+impl<'de> Deserialize<'de> for ChunkHash {
+    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
+        let s = String::deserialize(d)?;
+        ChunkHash::from_hex(&s).map_err(serde::de::Error::custom)
+    }
+}
+
+/// Summary of a stored blob without reading its contents.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BlobStat {
+    pub total_size: u64,
+    pub chunk_count: usize,
+}
+
+/// On-disk manifest for a blob. Public because the RPC layer (Phase 2b)
+/// serves this directly so a receiver can request only missing chunks.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BlobManifest {
+    pub blob_id: BlobId,
+    pub total_size: u64,
+    pub chunks: Vec<ChunkHash>,
+}
+
+/// Report from [`BlobStore::gc_orphan_chunks`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GcReport {
+    pub chunks_scanned: usize,
+    pub chunks_removed: usize,
+    pub bytes_reclaimed: u64,
+}
+
+/// Content-addressed blob store rooted at a filesystem directory.
+#[derive(Debug, Clone)]
+pub struct BlobStore {
+    root: PathBuf,
+}
+
+impl BlobStore {
+    /// Open (create if missing) a blob store rooted at `root`. Creates
+    /// the `blobs/`, `chunks/`, and `.tmp/` subdirs. Safe to call on
+    /// an existing store — no data is touched.
+    pub fn open(root: PathBuf) -> Result<Self> {
+        std::fs::create_dir_all(root.join("blobs"))
+            .with_context(|| format!("creating blobs dir under {}", root.display()))?;
+        std::fs::create_dir_all(root.join("chunks"))
+            .with_context(|| format!("creating chunks 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
+    }
+
+    /// Store `data`, returning the deterministic content hash.
+    /// Idempotent: writing the same bytes twice writes to the same
+    /// manifest path and yields the same BlobId. Chunks that already
+    /// exist on disk are not re-written.
+    pub async fn put_bytes(&self, data: &[u8]) -> Result<BlobId> {
+        let mut blob_hasher = blake3::Hasher::new();
+        blob_hasher.update(data);
+        let blob_id = BlobId(blob_hasher.finalize().into());
+
+        let mut chunk_hashes = Vec::with_capacity(data.len().div_ceil(CHUNK_SIZE));
+        for chunk in data.chunks(CHUNK_SIZE) {
+            let chunk_hash = ChunkHash(blake3::hash(chunk).into());
+            self.write_chunk_if_absent(&chunk_hash, chunk).await?;
+            chunk_hashes.push(chunk_hash);
+        }
+
+        let manifest = BlobManifest {
+            blob_id,
+            total_size: data.len() as u64,
+            chunks: chunk_hashes,
+        };
+        self.write_manifest_if_absent(&manifest).await?;
+        Ok(blob_id)
+    }
+
+    /// Read the entire blob back into memory. Returns `None` when no
+    /// manifest exists for `id`. Verifies the reconstructed bytes hash
+    /// back to `id` — mismatch means store corruption and returns Err.
+    pub async fn get_bytes(&self, id: &BlobId) -> Result<Option<Vec<u8>>> {
+        let manifest = match self.load_manifest(id).await? {
+            Some(m) => m,
+            None => return Ok(None),
+        };
+        let mut out = Vec::with_capacity(manifest.total_size as usize);
+        for chunk_hash in &manifest.chunks {
+            let path = self.chunk_path(chunk_hash);
+            let bytes = tokio::fs::read(&path)
+                .await
+                .with_context(|| format!("reading chunk {}", path.display()))?;
+            let recomputed = ChunkHash(blake3::hash(&bytes).into());
+            if recomputed != *chunk_hash {
+                bail!(
+                    "chunk hash mismatch at {}: manifest says {}, disk hashes to {}",
+                    path.display(),
+                    chunk_hash.to_hex(),
+                    recomputed.to_hex()
+                );
+            }
+            out.extend_from_slice(&bytes);
+        }
+        if out.len() as u64 != manifest.total_size {
+            bail!(
+                "reassembled blob size {} does not match manifest {}",
+                out.len(),
+                manifest.total_size
+            );
+        }
+        let final_hash = BlobId(blake3::hash(&out).into());
+        if final_hash != *id {
+            bail!(
+                "reassembled blob hash {} does not match requested {}",
+                final_hash.to_hex(),
+                id.to_hex()
+            );
+        }
+        Ok(Some(out))
+    }
+
+    /// Whether a manifest exists for `id`. Cheap — no chunk reads.
+    pub async fn contains(&self, id: &BlobId) -> Result<bool> {
+        match tokio::fs::metadata(self.manifest_path(id)).await {
+            Ok(_) => Ok(true),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+            Err(e) => Err(anyhow::Error::from(e)),
+        }
+    }
+
+    /// Metadata for a stored blob — size + chunk count — without reading
+    /// any chunk data.
+    pub async fn stat(&self, id: &BlobId) -> Result<Option<BlobStat>> {
+        Ok(self.load_manifest(id).await?.map(|m| BlobStat {
+            total_size: m.total_size,
+            chunk_count: m.chunks.len(),
+        }))
+    }
+
+    /// Load the raw manifest. Public so the RPC layer can serve it as
+    /// the response to a "list chunks I need" request.
+    pub async fn load_manifest(&self, id: &BlobId) -> Result<Option<BlobManifest>> {
+        let path = self.manifest_path(id);
+        let bytes = match tokio::fs::read(&path).await {
+            Ok(b) => b,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(e) => return Err(anyhow::Error::from(e)),
+        };
+        let manifest: BlobManifest = serde_json::from_slice(&bytes)
+            .with_context(|| format!("parsing manifest at {}", path.display()))?;
+        if manifest.blob_id != *id {
+            bail!(
+                "manifest at {} claims blob_id {} but was requested as {}",
+                path.display(),
+                manifest.blob_id.to_hex(),
+                id.to_hex()
+            );
+        }
+        Ok(Some(manifest))
+    }
+
+    /// Delete the top-level manifest for a blob. Chunks stay — they
+    /// may be shared with other blobs; [`gc_orphan_chunks`] does the
+    /// eventual reclamation.
+    ///
+    /// Returns `true` if a manifest was removed, `false` if none existed.
+    pub async fn delete_manifest(&self, id: &BlobId) -> Result<bool> {
+        let path = self.manifest_path(id);
+        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)),
+        }
+    }
+
+    /// Walk every live manifest to build the referenced-chunk set,
+    /// then remove any chunk file not in that set. Bounded by the
+    /// number of manifests + chunks on disk. Safe to interrupt: any
+    /// crash mid-sweep leaves live chunks intact (we only delete files
+    /// not in the referenced set).
+    pub async fn gc_orphan_chunks(&self) -> Result<GcReport> {
+        let referenced = self.collect_referenced_chunks().await?;
+
+        let chunks_root = self.root.join("chunks");
+        let mut chunks_scanned = 0usize;
+        let mut chunks_removed = 0usize;
+        let mut bytes_reclaimed = 0u64;
+
+        let mut top_entries = tokio::fs::read_dir(&chunks_root)
+            .await
+            .with_context(|| format!("reading {}", chunks_root.display()))?;
+        while let Some(bucket) = top_entries.next_entry().await? {
+            if !bucket.file_type().await?.is_dir() {
+                continue;
+            }
+            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
+            while let Some(entry) = inner.next_entry().await? {
+                if !entry.file_type().await?.is_file() {
+                    continue;
+                }
+                chunks_scanned += 1;
+                let name = entry.file_name();
+                let name_str = match name.to_str() {
+                    Some(s) => s,
+                    None => continue,
+                };
+                let hash = match ChunkHash::from_hex(name_str) {
+                    Ok(h) => h,
+                    Err(_) => continue, // unknown filename shape; leave alone
+                };
+                if referenced.contains(&hash) {
+                    continue;
+                }
+                if let Ok(meta) = entry.metadata().await {
+                    bytes_reclaimed = bytes_reclaimed.saturating_add(meta.len());
+                }
+                if tokio::fs::remove_file(entry.path()).await.is_ok() {
+                    chunks_removed += 1;
+                }
+            }
+        }
+
+        Ok(GcReport {
+            chunks_scanned,
+            chunks_removed,
+            bytes_reclaimed,
+        })
+    }
+
+    // ── internals ────────────────────────────────────────────────────
+
+    fn manifest_path(&self, id: &BlobId) -> PathBuf {
+        let hex = id.to_hex();
+        self.root
+            .join("blobs")
+            .join(&hex[..2])
+            .join(format!("{hex}.manifest.json"))
+    }
+
+    fn chunk_path(&self, hash: &ChunkHash) -> PathBuf {
+        let hex = hash.to_hex();
+        self.root.join("chunks").join(&hex[..2]).join(&hex)
+    }
+
+    /// Write a chunk if it isn't already present. Atomic via temp
+    /// file + rename so a concurrent reader never sees a partial file.
+    async fn write_chunk_if_absent(&self, hash: &ChunkHash, bytes: &[u8]) -> Result<()> {
+        let final_path = self.chunk_path(hash);
+        if tokio::fs::metadata(&final_path).await.is_ok() {
+            return Ok(());
+        }
+        if let Some(parent) = final_path.parent() {
+            tokio::fs::create_dir_all(parent).await.with_context(|| {
+                format!("creating chunk bucket {}", parent.display())
+            })?;
+        }
+        self.atomic_write(&final_path, bytes).await
+    }
+
+    /// Write manifest JSON if not already present. Same atomicity as
+    /// chunks; putting identical content is a no-op after the first.
+    async fn write_manifest_if_absent(&self, manifest: &BlobManifest) -> Result<()> {
+        let final_path = self.manifest_path(&manifest.blob_id);
+        if tokio::fs::metadata(&final_path).await.is_ok() {
+            return Ok(());
+        }
+        if let Some(parent) = final_path.parent() {
+            tokio::fs::create_dir_all(parent).await.with_context(|| {
+                format!("creating manifest bucket {}", parent.display())
+            })?;
+        }
+        let bytes = serde_json::to_vec_pretty(manifest)
+            .context("serialising manifest to JSON")?;
+        self.atomic_write(&final_path, &bytes).await
+    }
+
+    /// Write `bytes` to `final_path` atomically: stage in `.tmp/`, then
+    /// rename. Rename is atomic within a filesystem, and both source
+    /// and destination live under `self.root` so we're always on the
+    /// same FS.
+    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 file {}", 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(())
+    }
+
+    /// Walk every manifest and collect the union of chunk hashes they
+    /// reference. Used by [`gc_orphan_chunks`].
+    async fn collect_referenced_chunks(
+        &self,
+    ) -> Result<std::collections::HashSet<ChunkHash>> {
+        let blobs_root = self.root.join("blobs");
+        let mut referenced = std::collections::HashSet::new();
+        let mut top_entries = tokio::fs::read_dir(&blobs_root).await?;
+        while let Some(bucket) = top_entries.next_entry().await? {
+            if !bucket.file_type().await?.is_dir() {
+                continue;
+            }
+            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
+            while let Some(entry) = inner.next_entry().await? {
+                if !entry.file_type().await?.is_file() {
+                    continue;
+                }
+                let name = entry.file_name();
+                let name_str = match name.to_str() {
+                    Some(s) => s,
+                    None => continue,
+                };
+                if !name_str.ends_with(".manifest.json") {
+                    continue;
+                }
+                let bytes = tokio::fs::read(entry.path()).await?;
+                if let Ok(m) = serde_json::from_slice::<BlobManifest>(&bytes) {
+                    for c in m.chunks {
+                        referenced.insert(c);
+                    }
+                }
+            }
+        }
+        Ok(referenced)
+    }
+}
+
+/// Monotonic counter to disambiguate temp file names within a single
+/// process. Combined with the process id, this makes tmp filenames
+/// unique across a fleet without needing `Math.random`.
+static RANDOM_SUFFIX: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
+// ── tiny hex codec (no dep on `hex` crate) ────────────────────────────
+
+fn hex_encode(bytes: &[u8]) -> String {
+    let mut out = String::with_capacity(bytes.len() * 2);
+    for b in bytes {
+        out.push(nibble((b >> 4) & 0xf));
+        out.push(nibble(b & 0xf));
+    }
+    out
+}
+
+fn nibble(n: u8) -> char {
+    match n {
+        0..=9 => (b'0' + n) as char,
+        10..=15 => (b'a' + (n - 10)) as char,
+        _ => unreachable!("nibble is 4 bits"),
+    }
+}
+
+fn hex_decode(s: &str) -> Result<Vec<u8>> {
+    if s.len() % 2 != 0 {
+        bail!("hex input length must be even; got {}", s.len());
+    }
+    let bytes = s.as_bytes();
+    let mut out = Vec::with_capacity(s.len() / 2);
+    let mut i = 0;
+    while i < bytes.len() {
+        let hi = decode_nibble(bytes[i])?;
+        let lo = decode_nibble(bytes[i + 1])?;
+        out.push((hi << 4) | lo);
+        i += 2;
+    }
+    Ok(out)
+}
+
+fn decode_nibble(c: u8) -> Result<u8> {
+    match c {
+        b'0'..=b'9' => Ok(c - b'0'),
+        b'a'..=b'f' => Ok(c - b'a' + 10),
+        b'A'..=b'F' => Ok(c - b'A' + 10),
+        _ => bail!("non-hex character {:?}", c as char),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn open_store() -> (tempfile::TempDir, BlobStore) {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let store = BlobStore::open(tmp.path().to_path_buf()).unwrap();
+        (tmp, store)
+    }
+
+    #[test]
+    fn blob_id_hex_round_trips() {
+        let raw = [0xabu8; 32];
+        let id = BlobId::from_bytes(raw);
+        let hex = id.to_hex();
+        assert_eq!(hex.len(), 64);
+        assert_eq!(BlobId::from_hex(&hex).unwrap(), id);
+    }
+
+    #[test]
+    fn blob_id_from_hex_rejects_wrong_length() {
+        assert!(BlobId::from_hex("abcd").is_err());
+        assert!(BlobId::from_hex(&"a".repeat(63)).is_err());
+    }
+
+    #[test]
+    fn blob_id_from_hex_rejects_bad_chars() {
+        let bad: String = "z".repeat(64);
+        assert!(BlobId::from_hex(&bad).is_err());
+    }
+
+    #[test]
+    fn chunk_hash_hex_round_trips() {
+        let raw = [0x5cu8; 32];
+        let hash = ChunkHash::from_bytes(raw);
+        let hex = hash.to_hex();
+        assert_eq!(ChunkHash::from_hex(&hex).unwrap(), hash);
+    }
+
+    #[test]
+    fn blob_id_serde_uses_hex() {
+        let id = BlobId::from_bytes([0x33u8; 32]);
+        let json = serde_json::to_string(&id).unwrap();
+        assert_eq!(json, format!("\"{}\"", id.to_hex()));
+        let round: BlobId = serde_json::from_str(&json).unwrap();
+        assert_eq!(round, id);
+    }
+
+    #[tokio::test]
+    async fn open_creates_expected_layout() {
+        let (tmp, store) = open_store();
+        assert!(tmp.path().join("blobs").is_dir());
+        assert!(tmp.path().join("chunks").is_dir());
+        assert!(tmp.path().join(".tmp").is_dir());
+        assert_eq!(store.root(), tmp.path());
+    }
+
+    #[tokio::test]
+    async fn put_and_get_round_trips_small_blob() {
+        let (_tmp, store) = open_store();
+        let data = b"hello, clawstor!";
+        let id = store.put_bytes(data).await.unwrap();
+        // Deterministic: hashing the same bytes independently must
+        // produce the same BlobId.
+        let expected = BlobId::from_bytes(blake3::hash(data).into());
+        assert_eq!(id, expected);
+        let round = store.get_bytes(&id).await.unwrap();
+        assert_eq!(round.as_deref(), Some(data.as_slice()));
+    }
+
+    #[tokio::test]
+    async fn put_is_idempotent_and_deduplicates() {
+        let (_tmp, store) = open_store();
+        let data = vec![0xaau8; CHUNK_SIZE + 100];
+        let id1 = store.put_bytes(&data).await.unwrap();
+        let id2 = store.put_bytes(&data).await.unwrap();
+        assert_eq!(id1, id2, "same content → same BlobId");
+
+        // Two manifest writes for the same BlobId is fine; on disk we
+        // should still have exactly one manifest file.
+        let hex = id1.to_hex();
+        let manifest_dir = store.root().join("blobs").join(&hex[..2]);
+        let manifests: Vec<_> = std::fs::read_dir(&manifest_dir)
+            .unwrap()
+            .filter_map(|r| r.ok())
+            .collect();
+        assert_eq!(manifests.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn multi_chunk_blob_round_trips_and_matches_size() {
+        let (_tmp, store) = open_store();
+        // 10 MB payload — spans 3 chunks (4M + 4M + 2M).
+        let data: Vec<u8> = (0..10 * 1024 * 1024)
+            .map(|i| (i % 251) as u8)
+            .collect();
+        let id = store.put_bytes(&data).await.unwrap();
+        let stat = store.stat(&id).await.unwrap().unwrap();
+        assert_eq!(stat.total_size, data.len() as u64);
+        assert_eq!(stat.chunk_count, 3);
+        let round = store.get_bytes(&id).await.unwrap().unwrap();
+        assert_eq!(round.len(), data.len());
+        assert_eq!(round, data);
+    }
+
+    #[tokio::test]
+    async fn put_empty_blob_round_trips() {
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(&[]).await.unwrap();
+        let round = store.get_bytes(&id).await.unwrap().unwrap();
+        assert!(round.is_empty());
+        let stat = store.stat(&id).await.unwrap().unwrap();
+        assert_eq!(stat.total_size, 0);
+        assert_eq!(stat.chunk_count, 0);
+    }
+
+    #[tokio::test]
+    async fn different_content_yields_different_blob_ids() {
+        let (_tmp, store) = open_store();
+        let id_a = store.put_bytes(b"hello").await.unwrap();
+        let id_b = store.put_bytes(b"world").await.unwrap();
+        assert_ne!(id_a, id_b);
+    }
+
+    #[tokio::test]
+    async fn shared_chunks_are_stored_only_once() {
+        let (tmp, store) = open_store();
+        // Two blobs sharing the first CHUNK_SIZE bytes.
+        let mut a = vec![0x11u8; CHUNK_SIZE];
+        a.extend(vec![0x22u8; 100]);
+        let mut b = vec![0x11u8; CHUNK_SIZE];
+        b.extend(vec![0x33u8; 100]);
+        let id_a = store.put_bytes(&a).await.unwrap();
+        let id_b = store.put_bytes(&b).await.unwrap();
+        assert_ne!(id_a, id_b);
+
+        // Count unique chunks on disk. Both blobs share 1 chunk + each
+        // has 1 unique tail → 3 chunks total, not 4.
+        let chunk_count = count_files_under(&tmp.path().join("chunks"));
+        assert_eq!(chunk_count, 3, "shared chunk must be stored once");
+    }
+
+    #[tokio::test]
+    async fn get_returns_none_when_missing() {
+        let (_tmp, store) = open_store();
+        let missing = BlobId::from_bytes([0u8; 32]);
+        assert_eq!(store.get_bytes(&missing).await.unwrap(), None);
+        assert_eq!(store.stat(&missing).await.unwrap(), None);
+        assert!(!store.contains(&missing).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn contains_is_true_after_put() {
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(b"content").await.unwrap();
+        assert!(store.contains(&id).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn delete_manifest_removes_only_the_manifest() {
+        let (tmp, store) = open_store();
+        let id = store.put_bytes(b"once-lived-here").await.unwrap();
+        assert!(store.delete_manifest(&id).await.unwrap());
+        assert!(!store.contains(&id).await.unwrap());
+        // The chunk file must still exist — deletion is intentional
+        // orphaning, cleaned up by gc.
+        let chunk_count = count_files_under(&tmp.path().join("chunks"));
+        assert_eq!(chunk_count, 1, "chunk should persist after manifest delete");
+    }
+
+    #[tokio::test]
+    async fn delete_manifest_on_missing_returns_false() {
+        let (_tmp, store) = open_store();
+        let ghost = BlobId::from_bytes([0u8; 32]);
+        assert!(!store.delete_manifest(&ghost).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn gc_reclaims_orphan_chunks_but_keeps_referenced() {
+        let (tmp, store) = open_store();
+        // Blob A stays live; blob B gets its manifest deleted.
+        let id_a = store.put_bytes(b"blob-a-content").await.unwrap();
+        let id_b = store.put_bytes(b"blob-b-content").await.unwrap();
+        store.delete_manifest(&id_b).await.unwrap();
+
+        let before = count_files_under(&tmp.path().join("chunks"));
+        assert_eq!(before, 2, "both chunks should still be on disk pre-gc");
+
+        let report = store.gc_orphan_chunks().await.unwrap();
+        assert_eq!(report.chunks_scanned, 2);
+        assert_eq!(report.chunks_removed, 1);
+        assert!(report.bytes_reclaimed >= b"blob-b-content".len() as u64);
+
+        let after = count_files_under(&tmp.path().join("chunks"));
+        assert_eq!(after, 1, "only A's chunk should survive");
+        // A must still be readable.
+        let round = store.get_bytes(&id_a).await.unwrap().unwrap();
+        assert_eq!(round, b"blob-a-content");
+    }
+
+    #[tokio::test]
+    async fn gc_on_empty_store_reports_zero() {
+        let (_tmp, store) = open_store();
+        let report = store.gc_orphan_chunks().await.unwrap();
+        assert_eq!(report.chunks_scanned, 0);
+        assert_eq!(report.chunks_removed, 0);
+        assert_eq!(report.bytes_reclaimed, 0);
+    }
+
+    #[tokio::test]
+    async fn corrupted_chunk_detected_on_read() {
+        let (_tmp, store) = open_store();
+        let id = store.put_bytes(b"tamper-detectable").await.unwrap();
+        // Find + rewrite the single chunk file with garbage.
+        let hex = id.to_hex();
+        let bucket = store.root().join("chunks").join(&hex[..2]);
+        let entries: Vec<_> = std::fs::read_dir(&bucket)
+            .unwrap()
+            .filter_map(|r| r.ok())
+            .collect();
+        assert_eq!(entries.len(), 1);
+        std::fs::write(entries[0].path(), b"corrupted").unwrap();
+
+        let err = store.get_bytes(&id).await.unwrap_err().to_string();
+        assert!(
+            err.contains("chunk hash mismatch"),
+            "expected chunk mismatch error, got: {err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn load_manifest_returns_none_when_missing() {
+        let (_tmp, store) = open_store();
+        let missing = BlobId::from_bytes([0u8; 32]);
+        assert!(store.load_manifest(&missing).await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn load_manifest_round_trips_chunk_list() {
+        let (_tmp, store) = open_store();
+        let data = vec![0x42u8; CHUNK_SIZE + 1];
+        let id = store.put_bytes(&data).await.unwrap();
+        let m = store.load_manifest(&id).await.unwrap().unwrap();
+        assert_eq!(m.blob_id, id);
+        assert_eq!(m.total_size, data.len() as u64);
+        assert_eq!(m.chunks.len(), 2);
+    }
+
+    /// Recursive count of regular files under `root`. Test helper.
+    fn count_files_under(root: &Path) -> usize {
+        if !root.exists() {
+            return 0;
+        }
+        let mut total = 0;
+        let mut stack = vec![root.to_path_buf()];
+        while let Some(dir) = stack.pop() {
+            for entry in std::fs::read_dir(&dir).unwrap().flatten() {
+                let ft = entry.file_type().unwrap();
+                if ft.is_dir() {
+                    stack.push(entry.path());
+                } else if ft.is_file() {
+                    total += 1;
+                }
+            }
+        }
+        total
+    }
+}