claw-store/src/cluster.rs

@@ -20,6 +20,7 @@ pub mod client_config;
 pub mod gossip;
 pub mod metrics;
 pub mod prom;
+pub mod ref_tracking;
 pub mod refs;
 pub mod rpc;
 pub mod services;

claw-store/src/cluster/ref_tracking.rs

@@ -0,0 +1,489 @@
+//! Ref-tracking store (Phase 7f).
+//!
+//! Records which `(repo, git-ref)` combinations produced each
+//! fingerprint in the cache. Feeds a nightly deletion-eligibility
+//! sweep: fingerprints whose recorded refs are ALL gone from the
+//! upstream Gitea repo, AND whose `last_seen` is older than the
+//! configured retention window, are safe to evict.
+//!
+//! Why per-fingerprint (not per-blob):
+//! * Fingerprints are the cache keys claw-cargo uses. One fp maps
+//!   to one blob (the whole cache tarball). Tracking at the fp
+//!   layer keeps this store aligned with the claw-cargo boundary.
+//! * Blobs are content-addressed and may be shared. Ref-tracking
+//!   is about *why we kept this cache*, which is a per-fp concern.
+//!
+//! Layout:
+//!
+//! ```text
+//! <root>/ref-tracking/<hh>/<fp_hex>.json   # RefEntry, JSON
+//! ```
+//!
+//! On-disk format is JSON — human-readable + inspectable via `jq`.
+//! The store is small (one record per cached fingerprint) so JSON
+//! overhead is negligible.
+
+use anyhow::{bail, Context, Result};
+use serde::{Deserialize, Serialize};
+use std::collections::{HashMap, HashSet};
+use std::path::{Path, PathBuf};
+
+/// On-disk record for a single fingerprint's ref lineage.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RefEntry {
+    /// The fingerprint (32-byte hash from claw-cargo). Stored as
+    /// hex in the JSON.
+    #[serde(with = "hex32")]
+    pub fingerprint: [u8; 32],
+    /// Repo slug that produced this cache, e.g. `clawverse/clawstor`.
+    /// Same shape as the Gitea URL segment.
+    pub repo: String,
+    /// Git refs (branches + tags) known to have produced this
+    /// fingerprint. Never removed by `record` — the sweep decides
+    /// what's live.
+    pub refs: Vec<String>,
+    /// Wall-clock unix seconds of the first `record` call.
+    pub first_seen_unix: u64,
+    /// Wall-clock unix seconds of the most recent `record` call.
+    /// A fresh CI build touching an old fp updates this so recent
+    /// activity keeps it alive even if its refs are stale.
+    pub last_seen_unix: u64,
+}
+
+/// Filesystem-backed ref-tracking store rooted at a directory
+/// (typically the same directory the blob store lives under).
+#[derive(Debug, Clone)]
+pub struct RefTracking {
+    root: PathBuf,
+}
+
+impl RefTracking {
+    /// Open (create if missing) under `root`. Entries land in
+    /// `<root>/ref-tracking/`.
+    pub fn open(root: PathBuf) -> Result<Self> {
+        std::fs::create_dir_all(root.join("ref-tracking"))
+            .with_context(|| format!("creating ref-tracking dir under {}", root.display()))?;
+        Ok(Self { root })
+    }
+
+    /// Record a `(fingerprint, repo, git_ref)` observation.
+    ///
+    /// * First call for this fp: creates the entry with the single
+    ///   ref, both timestamps = `now_unix`.
+    /// * Subsequent calls: appends the ref if it's not already
+    ///   present, refreshes `last_seen_unix`. `repo` must match
+    ///   what's on disk — a fp may not be re-attributed to a
+    ///   different repo (that would silently mask a hash collision
+    ///   or a caller bug).
+    pub async fn record(
+        &self,
+        fingerprint: [u8; 32],
+        repo: &str,
+        git_ref: &str,
+        now_unix: u64,
+    ) -> Result<RefEntry> {
+        validate_repo(repo)?;
+        validate_ref(git_ref)?;
+        let path = self.entry_path(&fingerprint);
+        let mut entry = match self.load(&path).await? {
+            Some(existing) => {
+                if existing.repo != repo {
+                    bail!(
+                        "fingerprint {} already attributed to repo {:?}; cannot re-record under {:?}",
+                        hex32::encode(&fingerprint),
+                        existing.repo,
+                        repo
+                    );
+                }
+                existing
+            }
+            None => RefEntry {
+                fingerprint,
+                repo: repo.to_string(),
+                refs: Vec::new(),
+                first_seen_unix: now_unix,
+                last_seen_unix: now_unix,
+            },
+        };
+        if !entry.refs.iter().any(|r| r == git_ref) {
+            entry.refs.push(git_ref.to_string());
+        }
+        entry.last_seen_unix = now_unix;
+        // Keep refs stable across records so cross-node diff is
+        // easier.
+        entry.refs.sort();
+        entry.refs.dedup();
+        self.save(&path, &entry).await?;
+        Ok(entry)
+    }
+
+    /// Read the entry for a fingerprint, `None` if never recorded.
+    pub async fn get(&self, fingerprint: &[u8; 32]) -> Result<Option<RefEntry>> {
+        let path = self.entry_path(fingerprint);
+        self.load(&path).await
+    }
+
+    /// Enumerate every entry. Cost: one file read per entry.
+    /// Sorted by fingerprint for stable output.
+    pub async fn list_all(&self) -> Result<Vec<RefEntry>> {
+        let root = self.root.join("ref-tracking");
+        let mut out = Vec::new();
+        let mut top = match tokio::fs::read_dir(&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;
+                }
+                if entry.file_name().to_str().is_none_or(|n| !n.ends_with(".json")) {
+                    continue;
+                }
+                let bytes = match tokio::fs::read(entry.path()).await {
+                    Ok(b) => b,
+                    Err(_) => continue,
+                };
+                if let Ok(e) = serde_json::from_slice::<RefEntry>(&bytes) {
+                    out.push(e);
+                }
+            }
+        }
+        out.sort_by_key(|e| e.fingerprint);
+        Ok(out)
+    }
+
+    /// Deletion-eligibility sweep.
+    ///
+    /// A fingerprint is stale (returned) iff:
+    /// * every recorded ref is absent from `live_refs.get(&repo)`
+    ///   (or the repo has no live-refs entry at all — treated as
+    ///   "all refs gone")
+    /// * `now_unix - last_seen_unix >= retention_secs`
+    ///
+    /// `live_refs` is `{ repo_slug → set of live ref names }`,
+    /// typically fetched by the caller from Gitea just before the
+    /// sweep. Repos absent from the map are treated as fully gone.
+    ///
+    /// Returns the fingerprints. Callers apply the deletion (this
+    /// module never mutates anything but its own store).
+    pub async fn stale_at(
+        &self,
+        now_unix: u64,
+        live_refs: &HashMap<String, HashSet<String>>,
+        retention_secs: u64,
+    ) -> Result<Vec<[u8; 32]>> {
+        let mut out = Vec::new();
+        for entry in self.list_all().await? {
+            let age = now_unix.saturating_sub(entry.last_seen_unix);
+            if age < retention_secs {
+                continue;
+            }
+            let live_for_repo = live_refs.get(&entry.repo);
+            let all_dead = match live_for_repo {
+                None => true,
+                Some(live) => entry.refs.iter().all(|r| !live.contains(r)),
+            };
+            if all_dead {
+                out.push(entry.fingerprint);
+            }
+        }
+        Ok(out)
+    }
+
+    /// Remove one entry. Returns whether a file was actually removed.
+    /// Blob data is untouched — this only forgets the annotation.
+    pub async fn forget(&self, fingerprint: &[u8; 32]) -> Result<bool> {
+        let path = self.entry_path(fingerprint);
+        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 entry_path(&self, fingerprint: &[u8; 32]) -> PathBuf {
+        let hex = hex32::encode(fingerprint);
+        self.root
+            .join("ref-tracking")
+            .join(&hex[..2])
+            .join(format!("{hex}.json"))
+    }
+
+    async fn load(&self, path: &Path) -> Result<Option<RefEntry>> {
+        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 e: RefEntry = serde_json::from_slice(&bytes)
+            .with_context(|| format!("decoding {}", path.display()))?;
+        Ok(Some(e))
+    }
+
+    async fn save(&self, path: &Path, entry: &RefEntry) -> Result<()> {
+        use tokio::io::AsyncWriteExt;
+        let bytes = serde_json::to_vec_pretty(entry)
+            .context("serializing ref entry")?;
+        let parent = path.parent().context("entry path had no parent")?;
+        tokio::fs::create_dir_all(parent).await.with_context(|| {
+            format!("creating ref-tracking bucket {}", parent.display())
+        })?;
+        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 {}", tmp_path.display()))?;
+            f.write_all(&bytes).await?;
+            f.sync_all().await?;
+        }
+        tokio::fs::rename(&tmp_path, path)
+            .await
+            .with_context(|| {
+                format!("renaming {} → {}", tmp_path.display(), path.display())
+            })?;
+        Ok(())
+    }
+}
+
+static RANDOM_SUFFIX: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+fn validate_repo(repo: &str) -> Result<()> {
+    if repo.is_empty() {
+        bail!("repo cannot be empty");
+    }
+    if repo.len() > 1024 {
+        bail!("repo length {} exceeds cap", repo.len());
+    }
+    if repo.chars().any(|c| c.is_control() || c == '\0') {
+        bail!("repo has control character");
+    }
+    Ok(())
+}
+
+fn validate_ref(git_ref: &str) -> Result<()> {
+    if git_ref.is_empty() {
+        bail!("git ref cannot be empty");
+    }
+    if git_ref.len() > 1024 {
+        bail!("git ref length {} exceeds cap", git_ref.len());
+    }
+    if git_ref.chars().any(|c| c.is_control() || c == '\0') {
+        bail!("git ref has control character");
+    }
+    Ok(())
+}
+
+/// Hex-encoded 32-byte fingerprint for JSON serde. Kept in this
+/// file because it's the only place we need it and we want to
+/// avoid pulling in a wider hex-serde helper crate.
+mod hex32 {
+    use serde::{Deserialize, Deserializer, Serializer};
+
+    pub fn encode(bytes: &[u8; 32]) -> String {
+        let mut out = String::with_capacity(64);
+        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!(),
+        }
+    }
+
+    pub fn serialize<S: Serializer>(bytes: &[u8; 32], s: S) -> Result<S::Ok, S::Error> {
+        s.serialize_str(&encode(bytes))
+    }
+
+    pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 32], D::Error> {
+        let s = String::deserialize(d)?;
+        if s.len() != 64 {
+            return Err(serde::de::Error::custom(format!(
+                "expected 64-char hex, got {}",
+                s.len()
+            )));
+        }
+        let mut out = [0u8; 32];
+        for i in 0..32 {
+            let hi = decode_nibble(s.as_bytes()[i * 2])
+                .map_err(serde::de::Error::custom)?;
+            let lo = decode_nibble(s.as_bytes()[i * 2 + 1])
+                .map_err(serde::de::Error::custom)?;
+            out[i] = (hi << 4) | lo;
+        }
+        Ok(out)
+    }
+
+    fn decode_nibble(b: u8) -> Result<u8, String> {
+        match b {
+            b'0'..=b'9' => Ok(b - b'0'),
+            b'a'..=b'f' => Ok(b - b'a' + 10),
+            b'A'..=b'F' => Ok(b - b'A' + 10),
+            other => Err(format!("bad hex byte 0x{other:02x}")),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn open() -> (TempDir, RefTracking) {
+        let tmp = TempDir::new().unwrap();
+        let rt = RefTracking::open(tmp.path().to_path_buf()).unwrap();
+        (tmp, rt)
+    }
+
+    #[tokio::test]
+    async fn record_creates_entry_on_first_call() {
+        let (_tmp, rt) = open();
+        let fp = [0x11u8; 32];
+        let e = rt
+            .record(fp, "clawverse/clawstor", "main", 100)
+            .await
+            .unwrap();
+        assert_eq!(e.fingerprint, fp);
+        assert_eq!(e.repo, "clawverse/clawstor");
+        assert_eq!(e.refs, vec!["main"]);
+        assert_eq!(e.first_seen_unix, 100);
+        assert_eq!(e.last_seen_unix, 100);
+    }
+
+    #[tokio::test]
+    async fn record_appends_new_ref_and_refreshes_last_seen() {
+        let (_tmp, rt) = open();
+        let fp = [0x22u8; 32];
+        rt.record(fp, "r/x", "main", 100).await.unwrap();
+        let e2 = rt.record(fp, "r/x", "release/v2", 500).await.unwrap();
+        assert_eq!(e2.refs, vec!["main", "release/v2"]);
+        assert_eq!(e2.first_seen_unix, 100);
+        assert_eq!(e2.last_seen_unix, 500);
+    }
+
+    #[tokio::test]
+    async fn record_dedups_same_ref() {
+        let (_tmp, rt) = open();
+        let fp = [0x33u8; 32];
+        rt.record(fp, "r/x", "main", 100).await.unwrap();
+        let e = rt.record(fp, "r/x", "main", 200).await.unwrap();
+        assert_eq!(e.refs, vec!["main"]);
+        assert_eq!(e.last_seen_unix, 200);
+    }
+
+    #[tokio::test]
+    async fn record_rejects_repo_change() {
+        // A fp is deterministic from its inputs — the same fp
+        // showing up under two different repos means either a hash
+        // collision or a caller bug. Fail loud rather than silently
+        // re-attribute.
+        let (_tmp, rt) = open();
+        let fp = [0x44u8; 32];
+        rt.record(fp, "r/one", "main", 100).await.unwrap();
+        let err = rt.record(fp, "r/two", "main", 100).await.unwrap_err();
+        assert!(err.to_string().contains("r/one"));
+    }
+
+    #[tokio::test]
+    async fn stale_at_returns_fps_with_all_dead_refs_past_retention() {
+        // Setup: three fps, one live, one dead-but-fresh, one
+        // dead-and-old. Only the last should come back.
+        let (_tmp, rt) = open();
+        let live_fp = [0xA0u8; 32];
+        let recent_dead_fp = [0xA1u8; 32];
+        let old_dead_fp = [0xA2u8; 32];
+        rt.record(live_fp, "r/x", "main", 100).await.unwrap();
+        rt.record(recent_dead_fp, "r/x", "gone-branch", 900)
+            .await
+            .unwrap();
+        rt.record(old_dead_fp, "r/x", "another-gone-branch", 100)
+            .await
+            .unwrap();
+
+        let mut live = HashMap::new();
+        live.insert(
+            "r/x".to_string(),
+            ["main".to_string()].into_iter().collect(),
+        );
+        // now = 1000, retention = 500 seconds
+        let stale = rt.stale_at(1000, &live, 500).await.unwrap();
+        assert_eq!(stale, vec![old_dead_fp], "only aged + dead-refs qualifies");
+    }
+
+    #[tokio::test]
+    async fn stale_at_treats_missing_repo_entry_as_all_dead() {
+        // If Gitea has never heard of the repo (deleted repo, or
+        // sweep couldn't query it) → treat all refs as dead so we
+        // don't leak caches for gone repos.
+        let (_tmp, rt) = open();
+        let fp = [0xB0u8; 32];
+        rt.record(fp, "abandoned/repo", "main", 100)
+            .await
+            .unwrap();
+        let live = HashMap::new(); // repo not in map
+        let stale = rt.stale_at(1000, &live, 500).await.unwrap();
+        assert_eq!(stale, vec![fp]);
+    }
+
+    #[tokio::test]
+    async fn stale_at_respects_retention_window() {
+        // A fp with all refs dead but < retention_secs old must
+        // survive — retention protects fresh CI builds from being
+        // reaped before someone can rebuild against them.
+        let (_tmp, rt) = open();
+        let fp = [0xC0u8; 32];
+        rt.record(fp, "r/x", "dead-branch", 800).await.unwrap();
+        let live = HashMap::new();
+        // now = 1000, age = 200, retention = 500 → skip
+        let stale = rt.stale_at(1000, &live, 500).await.unwrap();
+        assert!(stale.is_empty());
+    }
+
+    #[tokio::test]
+    async fn forget_removes_entry_and_returns_truth() {
+        let (_tmp, rt) = open();
+        let fp = [0xD0u8; 32];
+        rt.record(fp, "r/x", "main", 100).await.unwrap();
+        assert!(rt.forget(&fp).await.unwrap());
+        assert!(rt.get(&fp).await.unwrap().is_none());
+        assert!(!rt.forget(&fp).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn list_all_sorted_by_fingerprint() {
+        let (_tmp, rt) = open();
+        rt.record([0x30; 32], "r/x", "main", 1).await.unwrap();
+        rt.record([0x10; 32], "r/x", "main", 1).await.unwrap();
+        rt.record([0x20; 32], "r/x", "main", 1).await.unwrap();
+        let all = rt.list_all().await.unwrap();
+        let fps: Vec<_> = all.iter().map(|e| e.fingerprint[0]).collect();
+        assert_eq!(fps, vec![0x10, 0x20, 0x30]);
+    }
+
+    #[tokio::test]
+    async fn validate_rejects_empty_and_control() {
+        assert!(validate_repo("").is_err());
+        assert!(validate_repo("with\0nul").is_err());
+        assert!(validate_repo("with\ncontrol").is_err());
+        assert!(validate_repo("ok/repo").is_ok());
+        assert!(validate_ref("").is_err());
+        assert!(validate_ref("refs/heads/main").is_ok());
+    }
+}