claw-store/src/cluster/services.rs

@@ -302,6 +302,14 @@ impl ClusterServices {
             (Some(store), Some(hours)) if hours > 0 => {
                 let store = store.clone();
                 let tag_store_for_gc = tag_store.clone();
+                // Phase 7d follow-on: snapshot store is under the
+                // same root as the blob store. Open once here so the
+                // ticker doesn't pay the fs setup cost every tick.
+                let snapshot_store_for_gc = blob_store_root
+                    .as_ref()
+                    .and_then(|root| {
+                        crate::cluster::snapshot::SnapshotStore::open(root.clone()).ok()
+                    });
                 let interval = Duration::from_secs(hours * 3600);
                 let max_gb = cluster.blob_max_gb;
                 Some(tokio::spawn(async move {
@@ -339,7 +347,7 @@ impl ClusterServices {
                                 .duration_since(std::time::UNIX_EPOCH)
                                 .map(|d| d.as_secs())
                                 .unwrap_or(0);
-                            let pinned = match &tag_store_for_gc {
+                            let mut pinned = match &tag_store_for_gc {
                                 Some(ts) => {
                                     // Phase 4b: prune expired pins first so
                                     // the pin set reflects the wall-clock
@@ -357,6 +365,17 @@ impl ClusterServices {
                                 }
                                 None => std::collections::HashSet::new(),
                             };
+                            // Phase 7d follow-on: snapshots pin their
+                            // referenced blobs. Union in every blob_id
+                            // captured by any snapshot; the eviction
+                            // routine sees the merged set.
+                            let mut snapshot_pin_count = 0usize;
+                            if let Some(ss) = &snapshot_store_for_gc {
+                                if let Ok(snaps) = ss.pinned_blob_ids().await {
+                                    snapshot_pin_count = snaps.len();
+                                    pinned.extend(snaps);
+                                }
+                            }
                             match store
                                 .evict_to_size_cap_with_pins(cap, &pinned)
                                 .await
@@ -366,6 +385,7 @@ impl ClusterServices {
                                     bytes_reclaimed = r.bytes_reclaimed,
                                     max_gb = gb,
                                     pinned_blobs = pinned.len(),
+                                    snapshot_pins = snapshot_pin_count,
                                     "auto-GC evicted LRU blobs to hit size cap"
                                 ),
                                 Ok(_) => {} // under cap already; keep quiet

claw-store/src/cluster/snapshot.rs

@@ -183,6 +183,28 @@ impl SnapshotStore {
         }
     }
 
+    /// Phase 7d follow-on: union of every `blob_id` referenced by ANY
+    /// snapshot. Feeds pin-aware eviction — every blob captured by a
+    /// live snapshot survives the LRU cap. Semantically "snapshots
+    /// act as immortal retention pins until the operator deletes
+    /// them".
+    ///
+    /// Cheap: one file read + JSON parse per snapshot. A fleet with
+    /// 100 snapshots × 10k blobs each is ~1 MB of JSON I/O.
+    pub async fn pinned_blob_ids(
+        &self,
+    ) -> Result<std::collections::HashSet<BlobId>> {
+        let mut out = std::collections::HashSet::new();
+        for summary in self.list().await? {
+            if let Some(m) = self.get(&summary.name).await? {
+                for id in m.blob_ids {
+                    out.insert(id);
+                }
+            }
+        }
+        Ok(out)
+    }
+
     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
@@ -347,6 +369,32 @@ mod tests {
         assert!(validate_name("v2026.07.14-pre-release").is_ok());
     }
 
+    #[tokio::test]
+    async fn pinned_blob_ids_unions_all_snapshots() {
+        // Two snapshots, some overlap. Union must dedupe.
+        let (_tmp, blob, snap) = open();
+        let a = blob.put_bytes(b"pin-alpha").await.unwrap();
+        let b = blob.put_bytes(b"pin-beta").await.unwrap();
+        let c = blob.put_bytes(b"pin-gamma").await.unwrap();
+        // Snapshot 1: captures a, b, c
+        snap.create("s1", &blob, 1).await.unwrap();
+        // Snapshot 2 (later): same content, still captures a, b, c
+        snap.create("s2", &blob, 2).await.unwrap();
+
+        let pins = snap.pinned_blob_ids().await.unwrap();
+        assert_eq!(pins.len(), 3);
+        assert!(pins.contains(&a));
+        assert!(pins.contains(&b));
+        assert!(pins.contains(&c));
+    }
+
+    #[tokio::test]
+    async fn pinned_blob_ids_empty_when_no_snapshots() {
+        let (_tmp, blob, snap) = open();
+        blob.put_bytes(b"blob-with-no-snapshot").await.unwrap();
+        assert!(snap.pinned_blob_ids().await.unwrap().is_empty());
+    }
+
     #[tokio::test]
     async fn round_trip_preserves_blob_ids_sorted() {
         // list_blob_ids order is filesystem-dependent. Snapshot

claw-store/src/main.rs

@@ -341,7 +341,7 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
         .duration_since(std::time::UNIX_EPOCH)
         .map(|d| d.as_secs())
         .unwrap_or(0);
-    let (pinned_blobs, expired_pruned) = if tags_dir.is_dir() {
+    let (mut pinned_blobs, expired_pruned) = if tags_dir.is_dir() {
         let ts = TagStore::open(tags_dir.clone())
             .with_context(|| format!("opening tag store at {}", tags_dir.display()))?;
         // Phase 4b: prune first so the set we hand to the evictor is
@@ -361,6 +361,18 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
     } else {
         (std::collections::HashSet::new(), 0)
     };
+    // Phase 7d follow-on: snapshot references act as immortal pins.
+    // Any blob captured by ANY snapshot survives the LRU cap so
+    // operators can guarantee retention windows via snapshots alone,
+    // without hand-managing per-blob tags.
+    let snapshot_store = cluster::snapshot::SnapshotStore::open(root.clone())
+        .context("opening snapshot store")?;
+    let snapshot_pins = snapshot_store
+        .pinned_blob_ids()
+        .await
+        .context("collecting snapshot pins")?;
+    let snapshot_pin_count = snapshot_pins.len();
+    pinned_blobs.extend(snapshot_pins);
 
     // Phase 2 (optional): pin-aware LRU eviction to hit a size cap.
     let evict = if let Some(gb) = evict_to_gb {
@@ -386,7 +398,7 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
         println!("lru eviction (cap {} GiB):", evict_to_gb.unwrap_or(0));
         println!("  chunks removed:  {}", evict.chunks_removed);
         println!("  bytes reclaimed: {}", evict.bytes_reclaimed);
-        println!("  pinned blobs:    {}", pinned_blobs.len());
+        println!("  pinned blobs:    {} ({} from snapshots)", pinned_blobs.len(), snapshot_pin_count);
         println!("  expired pins pruned: {}", expired_pruned);
     }
     println!("total elapsed:     {:?}", elapsed);