claw-store/src/cluster.rs

@@ -23,6 +23,7 @@ pub mod prom;
 pub mod refs;
 pub mod rpc;
 pub mod services;
+pub mod snapshot;
 pub mod tags;
 pub mod transport;
 pub mod wal;

claw-store/src/cluster/blob.rs

@@ -884,6 +884,45 @@ impl BlobStore {
         Ok(referenced)
     }
 
+    /// Phase 7d (2026-07-14): enumerate every blob currently in the
+    /// store. Cheap — reads only manifests, not chunk bodies. Used
+    /// by [`crate::cluster::snapshot::SnapshotStore::create`] to
+    /// build a point-in-time reference set. Order is filesystem walk
+    /// order — callers that need determinism must sort.
+    pub async fn list_blob_ids(&self) -> Result<Vec<BlobId>> {
+        let blobs_root = self.root.join("blobs");
+        let mut out = Vec::new();
+        let mut top = match tokio::fs::read_dir(&blobs_root).await {
+            Ok(e) => e,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
+            Err(e) => return Err(anyhow::Error::from(e)),
+        };
+        while let Some(bucket) = top.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,
+                };
+                let hex = match name_str.strip_suffix(".manifest.json") {
+                    Some(h) => h,
+                    None => continue,
+                };
+                if let Ok(id) = BlobId::from_hex(hex) {
+                    out.push(id);
+                }
+            }
+        }
+        Ok(out)
+    }
+
     /// Phase 7a (2026-07-14): read-only fsck for the blob store.
     ///
     /// For every `.manifest.json`, for every chunk it references:

claw-store/src/cluster/snapshot.rs

@@ -0,0 +1,366 @@
+//! Snapshot store (Phase 7d).
+//!
+//! A **snapshot** is a named, immutable record of every blob live in
+//! the store at the moment it was created. It is *not* a copy of the
+//! data — blobs are content-addressed and already live under
+//! `blobs/`. A snapshot is a list of `blob_id`s + wall-clock creation
+//! time, stored as JSON at
+//! `<root>/snapshots/<name>.json`.
+//!
+//! Why this exists:
+//! * **Rollback anchor** — before a risky migration, snapshot the
+//!   current graph; if things go sideways the snapshot lists exactly
+//!   which blobs must survive.
+//! * **Retention pin** — combined with the pin-aware LRU eviction
+//!   from Phase 4a, the operator can guarantee "these blobs stay on
+//!   disk for the next N days" without hand-listing them.
+//! * **Audit** — "which blobs existed at release time?"
+//!
+//! Layout:
+//!
+//! ```text
+//! <root>/snapshots/<name>.json     # SnapshotManifest, JSON
+//! ```
+//!
+//! Snapshot names are operator-supplied strings; the same
+//! `validate_name` rules that guard `TagStore` apply — printable
+//! non-slash characters, bounded length.
+
+use anyhow::{bail, Context, Result};
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+
+use crate::cluster::blob::{BlobId, BlobStore};
+
+/// Longest snapshot name we accept. Same rationale as
+/// [`crate::cluster::tags::MAX_TAG_KEY_BYTES`] but tighter — names
+/// end up in filesystem paths.
+pub const MAX_SNAPSHOT_NAME_BYTES: usize = 512;
+
+/// On-disk JSON representation of a snapshot.
+///
+/// `created_at_unix` is wall-clock seconds at the moment the snapshot
+/// was written. Stored explicitly so listing doesn't have to `stat`
+/// files, and so restore/report tools can render "created 3 days ago"
+/// without a filesystem call.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SnapshotManifest {
+    pub name: String,
+    pub created_at_unix: u64,
+    /// Blob ids that were live in the store at snapshot time.
+    /// Duplicates deduped by construction. Order is arbitrary.
+    pub blob_ids: Vec<BlobId>,
+}
+
+/// Compact form returned by [`SnapshotStore::list`]. Enough to render
+/// a table without slurping every snapshot's full blob list into
+/// memory.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SnapshotSummary {
+    pub name: String,
+    pub created_at_unix: u64,
+    pub blob_count: usize,
+    pub file_bytes: u64,
+}
+
+/// Filesystem-backed snapshot store rooted at a directory.
+#[derive(Debug, Clone)]
+pub struct SnapshotStore {
+    root: PathBuf,
+}
+
+impl SnapshotStore {
+    /// Open (create if missing) the snapshot store under `root`.
+    /// `root` is typically the same directory the blob store lives
+    /// under — snapshots land in `<root>/snapshots/`.
+    pub fn open(root: PathBuf) -> Result<Self> {
+        std::fs::create_dir_all(root.join("snapshots"))
+            .with_context(|| format!("creating snapshots dir under {}", root.display()))?;
+        Ok(Self { root })
+    }
+
+    /// Point-in-time snapshot of every blob currently in `store`.
+    /// Reads the live manifest set via `BlobStore::list_blob_ids`
+    /// and writes a single JSON file. Fails if a snapshot with the
+    /// same name already exists — snapshots are meant to be
+    /// immutable checkpoints, not mutable pointers. Use `delete`
+    /// then `create` if you really want to overwrite.
+    pub async fn create(
+        &self,
+        name: &str,
+        store: &BlobStore,
+        created_at_unix: u64,
+    ) -> Result<SnapshotManifest> {
+        validate_name(name)?;
+        let path = self.snapshot_path(name);
+        if tokio::fs::metadata(&path).await.is_ok() {
+            bail!("snapshot {:?} already exists at {}", name, path.display());
+        }
+        let mut blob_ids = store.list_blob_ids().await?;
+        // Stable order = stable JSON. Blob_ids from list_blob_ids come
+        // in filesystem walk order, which is not portable.
+        blob_ids.sort();
+        blob_ids.dedup();
+        let manifest = SnapshotManifest {
+            name: name.to_string(),
+            created_at_unix,
+            blob_ids,
+        };
+        let bytes = serde_json::to_vec_pretty(&manifest)
+            .context("serializing snapshot manifest")?;
+        atomic_write(&path, &bytes).await?;
+        Ok(manifest)
+    }
+
+    /// Read a snapshot back. `None` if unknown; `Err` if the file is
+    /// present but malformed (visibility over silent skip).
+    pub async fn get(&self, name: &str) -> Result<Option<SnapshotManifest>> {
+        validate_name(name)?;
+        let path = self.snapshot_path(name);
+        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 m: SnapshotManifest = serde_json::from_slice(&bytes)
+            .with_context(|| format!("decoding {}", path.display()))?;
+        Ok(Some(m))
+    }
+
+    /// Enumerate every snapshot. Cheap: one file read per snapshot
+    /// (the summary field is 3 numbers + a name). Returns entries
+    /// sorted by `created_at_unix` ascending — oldest first — so an
+    /// operator scanning a long list can find the oldest to prune.
+    pub async fn list(&self) -> Result<Vec<SnapshotSummary>> {
+        let dir = self.root.join("snapshots");
+        let mut out = Vec::new();
+        let mut entries = match tokio::fs::read_dir(&dir).await {
+            Ok(e) => e,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
+            Err(e) => return Err(anyhow::Error::from(e)),
+        };
+        while let Some(entry) = entries.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(".json") {
+                continue;
+            }
+            let bytes = match tokio::fs::read(entry.path()).await {
+                Ok(b) => b,
+                Err(_) => continue,
+            };
+            let file_bytes = bytes.len() as u64;
+            if let Ok(m) = serde_json::from_slice::<SnapshotManifest>(&bytes) {
+                out.push(SnapshotSummary {
+                    name: m.name,
+                    created_at_unix: m.created_at_unix,
+                    blob_count: m.blob_ids.len(),
+                    file_bytes,
+                });
+            }
+        }
+        out.sort_by_key(|s| s.created_at_unix);
+        Ok(out)
+    }
+
+    /// Remove a snapshot. Returns `true` if a file was removed,
+    /// `false` if none existed. Never touches blob storage —
+    /// deleting a snapshot only forgets the reference set, not the
+    /// blobs themselves.
+    pub async fn delete(&self, name: &str) -> Result<bool> {
+        validate_name(name)?;
+        let path = self.snapshot_path(name);
+        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)),
+        }
+    }
+
+    fn snapshot_path(&self, name: &str) -> PathBuf {
+        // Names are validated: no slashes, printable only. Safe to
+        // use directly as a filename fragment. We still normalize
+        // via hex-of-name if the operator ever passes something
+        // exotic like a unicode dash — but that's a future concern.
+        self.root
+            .join("snapshots")
+            .join(format!("{name}.json"))
+    }
+}
+
+fn validate_name(name: &str) -> Result<()> {
+    if name.is_empty() {
+        bail!("snapshot name cannot be empty");
+    }
+    if name.len() > MAX_SNAPSHOT_NAME_BYTES {
+        bail!(
+            "snapshot name length {} exceeds cap {}",
+            name.len(),
+            MAX_SNAPSHOT_NAME_BYTES
+        );
+    }
+    for c in name.chars() {
+        if c.is_control() {
+            bail!("snapshot name has control character");
+        }
+        if c == '/' || c == '\\' || c == '\0' {
+            bail!("snapshot name may not contain / \\ or NUL");
+        }
+    }
+    Ok(())
+}
+
+static RANDOM_SUFFIX: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+async fn atomic_write(final_path: &Path, bytes: &[u8]) -> Result<()> {
+    use tokio::io::AsyncWriteExt;
+    let parent = final_path
+        .parent()
+        .context("snapshot path had no parent")?;
+    let tmp_name = format!(
+        ".tmp.{}.{}",
+        std::process::id(),
+        RANDOM_SUFFIX.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+    );
+    let tmp_path = parent.join(tmp_name);
+    {
+        let mut f = tokio::fs::File::create(&tmp_path)
+            .await
+            .with_context(|| format!("creating tmp snapshot {}", 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(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn open() -> (TempDir, BlobStore, SnapshotStore) {
+        let tmp = TempDir::new().unwrap();
+        let blob = BlobStore::open(tmp.path().to_path_buf()).unwrap();
+        let snap = SnapshotStore::open(tmp.path().to_path_buf()).unwrap();
+        (tmp, blob, snap)
+    }
+
+    #[tokio::test]
+    async fn create_captures_all_live_blob_ids() {
+        // Three blobs in, snapshot should reference all three.
+        let (_tmp, blob, snap) = open();
+        let a = blob.put_bytes(b"alpha").await.unwrap();
+        let b = blob.put_bytes(b"beta").await.unwrap();
+        let c = blob.put_bytes(b"gamma").await.unwrap();
+        let m = snap.create("v1", &blob, 1_700_000_000).await.unwrap();
+        assert_eq!(m.name, "v1");
+        assert_eq!(m.created_at_unix, 1_700_000_000);
+        assert_eq!(m.blob_ids.len(), 3);
+        let ids: std::collections::HashSet<_> = m.blob_ids.iter().collect();
+        assert!(ids.contains(&a));
+        assert!(ids.contains(&b));
+        assert!(ids.contains(&c));
+    }
+
+    #[tokio::test]
+    async fn create_is_immutable_second_call_errors() {
+        // Snapshots are meant to be pin-in-time. Silent overwrite
+        // would be a footgun. Second create with same name must
+        // error, not clobber.
+        let (_tmp, blob, snap) = open();
+        blob.put_bytes(b"x").await.unwrap();
+        snap.create("v1", &blob, 1).await.unwrap();
+        let err = snap.create("v1", &blob, 2).await.unwrap_err();
+        assert!(err.to_string().contains("already exists"));
+    }
+
+    #[tokio::test]
+    async fn get_returns_none_for_missing() {
+        let (_tmp, _blob, snap) = open();
+        assert!(snap.get("nope").await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn list_sorts_by_creation_time_ascending() {
+        // Oldest first — operator triaging a growing list wants
+        // the pruning candidate at the top.
+        let (_tmp, blob, snap) = open();
+        blob.put_bytes(b"seed").await.unwrap();
+        snap.create("newer", &blob, 2000).await.unwrap();
+        snap.create("older", &blob, 1000).await.unwrap();
+        snap.create("newest", &blob, 3000).await.unwrap();
+        let entries = snap.list().await.unwrap();
+        let names: Vec<_> = entries.iter().map(|s| s.name.as_str()).collect();
+        assert_eq!(names, ["older", "newer", "newest"]);
+        assert_eq!(entries[0].blob_count, 1);
+        assert!(entries[0].file_bytes > 0);
+    }
+
+    #[tokio::test]
+    async fn delete_true_when_present_false_when_not() {
+        let (_tmp, blob, snap) = open();
+        blob.put_bytes(b"seed").await.unwrap();
+        snap.create("v1", &blob, 1).await.unwrap();
+        assert!(snap.delete("v1").await.unwrap());
+        assert!(!snap.delete("v1").await.unwrap());
+        assert!(snap.get("v1").await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn deleting_snapshot_does_not_touch_blobs() {
+        // A snapshot is a reference set, not a copy. Deleting one
+        // must not affect blob data — otherwise operators could
+        // accidentally nuke live data by pruning snapshots.
+        let (_tmp, blob, snap) = open();
+        let id = blob.put_bytes(b"survivor").await.unwrap();
+        snap.create("temp", &blob, 1).await.unwrap();
+        snap.delete("temp").await.unwrap();
+        // Blob still readable.
+        let back = blob.get_bytes(&id).await.unwrap();
+        assert_eq!(back.as_deref(), Some(&b"survivor"[..]));
+    }
+
+    #[tokio::test]
+    async fn validate_name_rejects_slash_and_control() {
+        assert!(validate_name("with/slash").is_err());
+        assert!(validate_name("with\\backslash").is_err());
+        assert!(validate_name("with\0null").is_err());
+        assert!(validate_name("with\x01ctrl").is_err());
+        assert!(validate_name("").is_err());
+        assert!(validate_name("ok-name").is_ok());
+        assert!(validate_name("v2026.07.14-pre-release").is_ok());
+    }
+
+    #[tokio::test]
+    async fn round_trip_preserves_blob_ids_sorted() {
+        // list_blob_ids order is filesystem-dependent. Snapshot
+        // must sort before writing so different nodes taking a
+        // snapshot of the same content get byte-identical files.
+        let (_tmp, blob, snap) = open();
+        for word in ["one", "two", "three", "four", "five"] {
+            blob.put_bytes(word.as_bytes()).await.unwrap();
+        }
+        let m = snap.create("v1", &blob, 42).await.unwrap();
+        let mut sorted = m.blob_ids.clone();
+        sorted.sort();
+        assert_eq!(m.blob_ids, sorted, "blob_ids must be sorted on write");
+        let m2 = snap.get("v1").await.unwrap().unwrap();
+        assert_eq!(m2.blob_ids, m.blob_ids);
+    }
+}

claw-store/src/main.rs

@@ -172,6 +172,29 @@ enum Cmd {
         #[arg(long)]
         evict_to_gb: Option<u64>,
     },
+    /// Phase 7d (2026-07-14): take a point-in-time snapshot of every
+    /// blob currently in the local store. Snapshots are cheap
+    /// reference sets (no data copy). Combine with pin-aware LRU
+    /// eviction to guarantee blobs stay on disk for a retention
+    /// window.
+    ClusterSnapshotCreate {
+        /// Operator-supplied snapshot name (no `/`, `\`, or NUL).
+        #[arg(long)]
+        name: String,
+    },
+    /// List every snapshot, oldest first.
+    ClusterSnapshotList,
+    /// Show a snapshot's full blob-id list.
+    ClusterSnapshotShow {
+        #[arg(long)]
+        name: String,
+    },
+    /// Remove a snapshot. Does NOT touch the referenced blob data —
+    /// snapshots are pointer-sets, not copies.
+    ClusterSnapshotDelete {
+        #[arg(long)]
+        name: String,
+    },
     /// Phase 7c (2026-07-14): fix corrupt/missing chunks by pulling
     /// them from a peer. Runs scrub first; if nothing bad, exits
     /// clean. Otherwise probes the peer (HasChunk) for each unique
@@ -268,6 +291,10 @@ async fn main() -> Result<()> {
             tls_dir,
             dry_run,
         } => cmd_cluster_repair(&cfg, &peer, rpc_addr, &tls_dir, dry_run).await?,
+        Cmd::ClusterSnapshotCreate { name } => cmd_cluster_snapshot_create(&cfg, &name).await?,
+        Cmd::ClusterSnapshotList => cmd_cluster_snapshot_list(&cfg).await?,
+        Cmd::ClusterSnapshotShow { name } => cmd_cluster_snapshot_show(&cfg, &name).await?,
+        Cmd::ClusterSnapshotDelete { name } => cmd_cluster_snapshot_delete(&cfg, &name).await?,
         Cmd::ClusterPeerStatus {
             peer,
             rpc_addr,
@@ -552,6 +579,92 @@ async fn cmd_cluster_repair(
     Ok(())
 }
 
+fn open_blob_and_snapshot_stores(
+    cfg: &Config,
+) -> Result<(cluster::blob::BlobStore, cluster::snapshot::SnapshotStore)> {
+    let root = cfg
+        .cluster
+        .as_ref()
+        .and_then(|c| c.blob_store_root.clone())
+        .context("cluster.blob_store_root not configured")?;
+    if !root.is_dir() {
+        bail!("blob_store_root {} does not exist", root.display());
+    }
+    let blob = cluster::blob::BlobStore::open(root.clone())
+        .with_context(|| format!("opening blob store at {}", root.display()))?;
+    let snap = cluster::snapshot::SnapshotStore::open(root.clone())
+        .with_context(|| format!("opening snapshot store at {}", root.display()))?;
+    Ok((blob, snap))
+}
+
+async fn cmd_cluster_snapshot_create(cfg: &Config, name: &str) -> Result<()> {
+    let (blob, snap) = open_blob_and_snapshot_stores(cfg)?;
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0);
+    let started = std::time::Instant::now();
+    let m = snap.create(name, &blob, now).await?;
+    println!("── clawstor snapshot-create ────────────────────────");
+    println!("name:              {}", m.name);
+    println!("created_at (unix): {}", m.created_at_unix);
+    println!("blob count:        {}", m.blob_ids.len());
+    println!("elapsed:           {:?}", started.elapsed());
+    println!("────────────────────────────────────────────────────");
+    Ok(())
+}
+
+async fn cmd_cluster_snapshot_list(cfg: &Config) -> Result<()> {
+    let (_blob, snap) = open_blob_and_snapshot_stores(cfg)?;
+    let entries = snap.list().await?;
+    println!("── clawstor snapshots ──────────────────────────────");
+    if entries.is_empty() {
+        println!("(no snapshots)");
+    } else {
+        println!(
+            "{:<20} {:<20} {:<10} {}",
+            "CREATED_AT", "NAME", "BLOBS", "SIZE"
+        );
+        for s in &entries {
+            println!(
+                "{:<20} {:<20} {:<10} {}",
+                s.created_at_unix, s.name, s.blob_count, s.file_bytes
+            );
+        }
+    }
+    println!("────────────────────────────────────────────────────");
+    Ok(())
+}
+
+async fn cmd_cluster_snapshot_show(cfg: &Config, name: &str) -> Result<()> {
+    let (_blob, snap) = open_blob_and_snapshot_stores(cfg)?;
+    let m = snap
+        .get(name)
+        .await?
+        .with_context(|| format!("snapshot {:?} not found", name))?;
+    println!("── clawstor snapshot show ──────────────────────────");
+    println!("name:              {}", m.name);
+    println!("created_at (unix): {}", m.created_at_unix);
+    println!("blob count:        {}", m.blob_ids.len());
+    println!("blob ids:");
+    for id in &m.blob_ids {
+        println!("  {}", id.to_hex());
+    }
+    println!("────────────────────────────────────────────────────");
+    Ok(())
+}
+
+async fn cmd_cluster_snapshot_delete(cfg: &Config, name: &str) -> Result<()> {
+    let (_blob, snap) = open_blob_and_snapshot_stores(cfg)?;
+    let removed = snap.delete(name).await?;
+    if removed {
+        println!("snapshot {:?} deleted (blob data untouched)", name);
+    } else {
+        println!("snapshot {:?} did not exist", name);
+    }
+    Ok(())
+}
+
 async fn cmd_cluster_peer_status(
     peer: &str,
     rpc_addr: SocketAddr,