claw-store/src/cluster/services.rs

@@ -335,9 +335,22 @@ impl ClusterServices {
                             // markers so operators can `claw-cargo pin`
                             // a build and know it won't be evicted by
                             // the size cap.
+                            let now_unix = std::time::SystemTime::now()
+                                .duration_since(std::time::UNIX_EPOCH)
+                                .map(|d| d.as_secs())
+                                .unwrap_or(0);
                             let pinned = match &tag_store_for_gc {
                                 Some(ts) => {
-                                    let raw = ts.pinned_blob_values().await.unwrap_or_default();
+                                    // Phase 4b: prune expired pins first so
+                                    // the pin set reflects the wall-clock
+                                    // moment we're about to evict at.
+                                    let _ = ts
+                                        .prune_expired_stamped_at(now_unix)
+                                        .await;
+                                    let raw = ts
+                                        .pinned_blob_values_at(now_unix)
+                                        .await
+                                        .unwrap_or_default();
                                     raw.into_iter()
                                         .map(crate::cluster::blob::BlobId::from_bytes)
                                         .collect()

claw-store/src/cluster/tags.rs

@@ -277,6 +277,20 @@ impl TagStore {
     /// is well under 100 KiB of memory.
     pub async fn pinned_blob_values(
         &self,
+    ) -> Result<std::collections::HashSet<[u8; 32]>> {
+        // No `now` filter → treat as "no expiry gate" by passing u64::MAX,
+        // so any tag whose expiry ≤ u64::MAX (i.e. any) still counts. All
+        // expiry-aware callers should use `pinned_blob_values_at`.
+        self.pinned_blob_values_at(u64::MAX).await
+    }
+
+    /// Phase 4b (2026-07-13): expiry-aware pin gathering. A stamped tag
+    /// is treated as pinning its value only if either (a) it has no
+    /// sidecar expiry, or (b) `expires_at > now`. Legacy `tags/` entries
+    /// have no expiry surface and always count.
+    pub async fn pinned_blob_values_at(
+        &self,
+        now_unix: u64,
     ) -> Result<std::collections::HashSet<[u8; 32]>> {
         let mut out = std::collections::HashSet::new();
         // Unstamped: same walk as `list`, but we skip TagEntry hex
@@ -321,7 +335,23 @@ impl TagStore {
                         Err(_) => continue,
                     };
                     if let Ok((_, stamped)) = decode_stamped_record(&bytes) {
-                        out.insert(stamped.value);
+                        // Check for `<path>.exp` sidecar. Absent → no
+                        // expiry. Present + not-yet-expired → still counts.
+                        // Present + expired → drop.
+                        let mut exp_path = entry.path();
+                        exp_path.set_extension("svtag.exp");
+                        let keep = match tokio::fs::read(&exp_path).await {
+                            Ok(bytes) if bytes.len() == 8 => {
+                                let expires_at = u64::from_le_bytes(
+                                    bytes.as_slice().try_into().unwrap(),
+                                );
+                                expires_at > now_unix
+                            }
+                            _ => true,
+                        };
+                        if keep {
+                            out.insert(stamped.value);
+                        }
                     }
                 }
             }
@@ -329,6 +359,95 @@ impl TagStore {
         Ok(out)
     }
 
+    /// Phase 4b: attach an expiry to a stamped tag. `expires_at_unix`
+    /// is absolute wall-clock seconds; `0` means "never expire" and
+    /// clears any existing sidecar. Writing to a key that has no
+    /// stamped tag on disk is not an error — the sidecar is written
+    /// and will start filtering once the tag lands.
+    pub async fn set_stamped_expiry(&self, key: &str, expires_at_unix: u64) -> Result<()> {
+        validate_key(key)?;
+        let path = self.stamped_expiry_path(key);
+        if expires_at_unix == 0 {
+            match tokio::fs::remove_file(&path).await {
+                Ok(()) => {}
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+                Err(e) => return Err(anyhow::Error::from(e)),
+            }
+            return Ok(());
+        }
+        if let Some(parent) = path.parent() {
+            tokio::fs::create_dir_all(parent).await.with_context(|| {
+                format!("creating stamped-tag bucket {}", parent.display())
+            })?;
+        }
+        self.atomic_write(&path, &expires_at_unix.to_le_bytes()).await
+    }
+
+    /// Phase 4b: read the expiry sidecar for a stamped tag, if any.
+    pub async fn get_stamped_expiry(&self, key: &str) -> Result<Option<u64>> {
+        validate_key(key)?;
+        let path = self.stamped_expiry_path(key);
+        match tokio::fs::read(&path).await {
+            Ok(bytes) if bytes.len() == 8 => {
+                Ok(Some(u64::from_le_bytes(bytes.as_slice().try_into().unwrap())))
+            }
+            Ok(_) => bail!("stamped-tag expiry sidecar at {} has wrong length", path.display()),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(e) => Err(anyhow::Error::from(e)),
+        }
+    }
+
+    /// Phase 4b: delete expired stamped tags and their sidecars.
+    /// Returns count removed. A tag is expired iff its `.svtag.exp`
+    /// sidecar contains an `expires_at ≤ now_unix`. Legacy `tags/`
+    /// entries are never touched here — they have no expiry surface.
+    pub async fn prune_expired_stamped_at(&self, now_unix: u64) -> Result<usize> {
+        let mut removed = 0usize;
+        let stamped_root = self.root.join("tags-v2");
+        if !stamped_root.is_dir() {
+            return Ok(0);
+        }
+        let mut top = tokio::fs::read_dir(&stamped_root).await?;
+        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 tag_path = entry.path();
+                if tag_path.extension().and_then(|s| s.to_str()) != Some("svtag") {
+                    continue;
+                }
+                let mut exp_path = tag_path.clone();
+                exp_path.set_extension("svtag.exp");
+                let expires_at = match tokio::fs::read(&exp_path).await {
+                    Ok(bytes) if bytes.len() == 8 => {
+                        u64::from_le_bytes(bytes.as_slice().try_into().unwrap())
+                    }
+                    _ => continue,
+                };
+                if expires_at > now_unix {
+                    continue;
+                }
+                let _ = tokio::fs::remove_file(&tag_path).await;
+                let _ = tokio::fs::remove_file(&exp_path).await;
+                removed += 1;
+            }
+        }
+        Ok(removed)
+    }
+
+    fn stamped_expiry_path(&self, key: &str) -> PathBuf {
+        let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
+        self.root
+            .join("tags-v2")
+            .join(&hash_hex[..2])
+            .join(format!("{hash_hex}.svtag.exp"))
+    }
+
     fn tag_path(&self, key: &str) -> PathBuf {
         let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
         self.root
@@ -755,6 +874,52 @@ mod tests {
         assert_eq!(pins.len(), 2, "duplicate values dedupe");
     }
 
+    #[tokio::test]
+    async fn expiry_gates_pin_set_and_prune_removes_expired() {
+        // Phase 4b: sidecar `.exp` files scope the pin protection
+        // window. A stamped tag with expires_at ≤ now is neither
+        // reported as a pinned value nor should it survive
+        // `prune_expired_stamped_at(now)`.
+        let (_tmp, store) = open();
+
+        let live = StampedTagValue { value: [0x01; 32], clock: 1, node: [0; 8] };
+        let expired = StampedTagValue { value: [0x02; 32], clock: 1, node: [0; 8] };
+        let no_ttl = StampedTagValue { value: [0x03; 32], clock: 1, node: [0; 8] };
+        store.put_stamped("live", live).await.unwrap();
+        store.put_stamped("expired", expired).await.unwrap();
+        store.put_stamped("no-ttl", no_ttl).await.unwrap();
+
+        // now = 1000. live expires at 2000, expired at 500.
+        store.set_stamped_expiry("live", 2000).await.unwrap();
+        store.set_stamped_expiry("expired", 500).await.unwrap();
+        // no-ttl deliberately has no sidecar.
+
+        // Round-trip the sidecar.
+        assert_eq!(store.get_stamped_expiry("live").await.unwrap(), Some(2000));
+        assert_eq!(store.get_stamped_expiry("no-ttl").await.unwrap(), None);
+
+        // Pin set at now=1000: only live + no-ttl.
+        let pins = store.pinned_blob_values_at(1000).await.unwrap();
+        assert!(pins.contains(&[0x01; 32]));
+        assert!(pins.contains(&[0x03; 32]));
+        assert!(!pins.contains(&[0x02; 32]), "expired pin must not count");
+        assert_eq!(pins.len(), 2);
+
+        // Prune at now=1000: only "expired" goes.
+        let removed = store.prune_expired_stamped_at(1000).await.unwrap();
+        assert_eq!(removed, 1);
+        assert_eq!(store.get_stamped("expired").await.unwrap(), None);
+        assert!(store.get_stamped("live").await.unwrap().is_some());
+        assert!(store.get_stamped("no-ttl").await.unwrap().is_some());
+
+        // Clear via expires_at=0 removes the sidecar.
+        store.set_stamped_expiry("live", 0).await.unwrap();
+        assert_eq!(store.get_stamped_expiry("live").await.unwrap(), None);
+        // Now "live" pin has no expiry gate → always counts.
+        let pins_after = store.pinned_blob_values_at(0).await.unwrap();
+        assert!(pins_after.contains(&[0x01; 32]));
+    }
+
     #[tokio::test]
     async fn stamped_and_unstamped_stores_are_independent() {
         // put() writes to tags/, put_stamped() writes to tags-v2/.

claw-store/src/main.rs

@@ -272,17 +272,29 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
     // respect them. Cheap even at fleet scale (one 32-byte value per
     // tag).
     let tags_dir = root.join("tags-db");
-    let pinned_blobs = if tags_dir.is_dir() {
+    let now_unix = std::time::SystemTime::now()
+        .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 ts = TagStore::open(tags_dir.clone())
             .with_context(|| format!("opening tag store at {}", tags_dir.display()))?;
-        ts.pinned_blob_values()
+        // Phase 4b: prune first so the set we hand to the evictor is
+        // current as of `now_unix`.
+        let pruned = ts
+            .prune_expired_stamped_at(now_unix)
+            .await
+            .context("pruning expired pins")?;
+        let pins = ts
+            .pinned_blob_values_at(now_unix)
             .await
             .context("collecting pinned tag values")?
             .into_iter()
             .map(BlobId::from_bytes)
-            .collect::<std::collections::HashSet<_>>()
+            .collect::<std::collections::HashSet<_>>();
+        (pins, pruned)
     } else {
-        std::collections::HashSet::new()
+        (std::collections::HashSet::new(), 0)
     };
 
     // Phase 2 (optional): pin-aware LRU eviction to hit a size cap.
@@ -310,6 +322,7 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
         println!("  chunks removed:  {}", evict.chunks_removed);
         println!("  bytes reclaimed: {}", evict.bytes_reclaimed);
         println!("  pinned blobs:    {}", pinned_blobs.len());
+        println!("  expired pins pruned: {}", expired_pruned);
     }
     println!("total elapsed:     {:?}", elapsed);
     println!("────────────────────────────────────────────────────");