Phase 9 R1: RepoEnsure — peer RPC + aggregator fan-out (#106)

claw-store/src/cluster/repo_ensure.rs

@@ -0,0 +1,335 @@
+//! Phase 9 R1a: peer-side git materialization.
+//!
+//! Two RPCs give the aggregator (or any authorized client) a way to
+//! ensure a `(url, git_ref)` pair is checked out on this node under a
+//! caller-provided workspace namespace, and to release it later.
+//!
+//! Nothing here fans out to peers — the aggregator layer is
+//! responsible for calling every node. This module is intentionally
+//! narrow: one node, one path derivation, one shallow clone.
+
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+use tokio::process::Command;
+use tokio::time::timeout;
+
+/// Cap on how long a clone can take. Shallow clones of even a large
+/// repo over LAN complete in seconds; anything past 5 minutes is a
+/// stuck network or a pathologically large tree.
+const CLONE_TIMEOUT: Duration = Duration::from_secs(300);
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoEnsureRequest {
+    pub url: String,
+    pub git_ref: String,
+    pub workspace: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoEnsureReply {
+    pub path: String,
+    pub head_sha: String,
+    /// True when the checkout was already present with a valid `.git`
+    /// and no reclone was needed.
+    pub cached: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoReleaseRequest {
+    pub url: String,
+    pub git_ref: String,
+    pub workspace: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoReleaseReply {
+    /// True when the path existed and was removed. False when nothing
+    /// was on disk to begin with (still success from the caller's POV).
+    pub removed: bool,
+}
+
+/// Sanitize an untrusted path component so it can never escape its
+/// parent. Replaces `/`, `\`, `..`, control characters, and leading
+/// dots. The result is always non-empty and safe to `join` under a
+/// known root.
+fn sanitize_component(input: &str) -> String {
+    if input.is_empty() {
+        return "_".to_string();
+    }
+    let mut out = String::with_capacity(input.len());
+    for ch in input.chars() {
+        let mapped = match ch {
+            '/' | '\\' | ':' | '\0' => '_',
+            c if c.is_control() => '_',
+            c => c,
+        };
+        out.push(mapped);
+    }
+    // Reject traversal — after char-mapping we could still have "..".
+    // Collapse any run of dots at either end to `_` prefix.
+    let trimmed = out.trim_matches('.');
+    if trimmed.is_empty() {
+        return "_".to_string();
+    }
+    // Replace embedded ".." segments defensively.
+    trimmed.replace("..", "__")
+}
+
+/// Deterministic on-disk path for a `(workspace, url, git_ref)` triple.
+/// The URL is hashed so we don't leak credentials or full URLs into
+/// directory names; the ref is sanitized so branch names with slashes
+/// (`feature/x`) don't create nested dirs.
+pub fn derive_path(repo_root: &Path, workspace: &str, url: &str, git_ref: &str) -> PathBuf {
+    let ws = sanitize_component(workspace);
+    let ref_slug = sanitize_component(git_ref);
+    let hex = blake3::hash(url.as_bytes()).to_hex();
+    let leaf = format!("{}-{}", &hex.as_str()[..16], ref_slug);
+    repo_root.join(ws).join(leaf)
+}
+
+/// Run `git -C <path> rev-parse HEAD` and return the resulting sha
+/// as a lowercase hex string. `Err` on any failure (including path
+/// not being a git repo).
+async fn read_head_sha(path: &Path) -> Result<String> {
+    let output = Command::new("git")
+        .arg("-C")
+        .arg(path)
+        .args(["rev-parse", "HEAD"])
+        .output()
+        .await
+        .context("spawning `git rev-parse`")?;
+    if !output.status.success() {
+        anyhow::bail!(
+            "git rev-parse failed at {}: {}",
+            path.display(),
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
+    let sha = String::from_utf8(output.stdout)
+        .context("git rev-parse output was not utf-8")?
+        .trim()
+        .to_string();
+    Ok(sha)
+}
+
+/// Ensure `(url, git_ref)` is materialized under `repo_root`. When a
+/// prior checkout already exists and its `.git` resolves cleanly, we
+/// treat that as a cache hit and return without touching disk. On any
+/// mismatch we remove-and-reclone.
+pub async fn ensure_repo(repo_root: &Path, req: &RepoEnsureRequest) -> Result<RepoEnsureReply> {
+    let path = derive_path(repo_root, &req.workspace, &req.url, &req.git_ref);
+
+    if path.join(".git").exists() {
+        if let Ok(sha) = read_head_sha(&path).await {
+            return Ok(RepoEnsureReply {
+                path: path.display().to_string(),
+                head_sha: sha,
+                cached: true,
+            });
+        }
+        // .git present but rev-parse failed — treat as corrupt and
+        // reclone.
+    }
+
+    // Ensure any partial prior attempt is cleared before we clone.
+    if path.exists() {
+        tokio::fs::remove_dir_all(&path)
+            .await
+            .with_context(|| format!("removing stale checkout at {}", path.display()))?;
+    }
+    if let Some(parent) = path.parent() {
+        tokio::fs::create_dir_all(parent)
+            .await
+            .with_context(|| format!("creating parent {}", parent.display()))?;
+    }
+
+    let clone = Command::new("git")
+        .args(["clone", "--depth", "1", "--branch", &req.git_ref])
+        .arg(&req.url)
+        .arg(&path)
+        .output();
+
+    let output = timeout(CLONE_TIMEOUT, clone)
+        .await
+        .with_context(|| format!("git clone timed out after {:?}", CLONE_TIMEOUT))?
+        .context("spawning `git clone`")?;
+
+    if !output.status.success() {
+        // Best-effort cleanup so we don't leave a half-clone behind.
+        let _ = tokio::fs::remove_dir_all(&path).await;
+        anyhow::bail!(
+            "git clone failed for {}@{}: {}",
+            req.url,
+            req.git_ref,
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
+
+    let sha = read_head_sha(&path).await?;
+    Ok(RepoEnsureReply {
+        path: path.display().to_string(),
+        head_sha: sha,
+        cached: false,
+    })
+}
+
+/// Remove the on-disk checkout for `(url, git_ref)` under `repo_root`.
+/// A missing path is not an error — the caller's precondition
+/// (nothing at this key) is already satisfied.
+pub async fn release_repo(
+    repo_root: &Path,
+    req: &RepoReleaseRequest,
+) -> Result<RepoReleaseReply> {
+    let path = derive_path(repo_root, &req.workspace, &req.url, &req.git_ref);
+    if !path.exists() {
+        return Ok(RepoReleaseReply { removed: false });
+    }
+    tokio::fs::remove_dir_all(&path)
+        .await
+        .with_context(|| format!("removing checkout at {}", path.display()))?;
+    Ok(RepoReleaseReply { removed: true })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn derive_path_is_deterministic() {
+        let root = PathBuf::from("/tmp/repos");
+        let a = derive_path(&root, "workspace:abc", "https://example.com/x.git", "main");
+        let b = derive_path(&root, "workspace:abc", "https://example.com/x.git", "main");
+        assert_eq!(a, b);
+    }
+
+    #[test]
+    fn derive_path_differs_by_ref() {
+        let root = PathBuf::from("/tmp/repos");
+        let a = derive_path(&root, "ws", "https://example.com/x.git", "main");
+        let b = derive_path(&root, "ws", "https://example.com/x.git", "develop");
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn derive_path_differs_by_url() {
+        let root = PathBuf::from("/tmp/repos");
+        let a = derive_path(&root, "ws", "https://example.com/x.git", "main");
+        let b = derive_path(&root, "ws", "https://example.com/y.git", "main");
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn derive_path_traversal_safe() {
+        let root = PathBuf::from("/tmp/repos");
+        // Malicious workspace tries to escape the root.
+        let p = derive_path(&root, "../etc", "https://example.com/x.git", "main");
+        assert!(
+            p.starts_with(&root),
+            "sanitized workspace must stay under root: got {}",
+            p.display()
+        );
+        assert!(!p.to_string_lossy().contains(".."));
+    }
+
+    #[test]
+    fn derive_path_slashy_ref_is_flat() {
+        let root = PathBuf::from("/tmp/repos");
+        let p = derive_path(&root, "ws", "https://example.com/x.git", "feature/nested");
+        // Ref becomes part of a single filename, not a nested dir.
+        assert_eq!(p.parent().unwrap().parent().unwrap(), root.as_path());
+    }
+
+    /// End-to-end: seed a bare git repo in a tempdir, ensure it into
+    /// a fresh repo_root, verify cached=true on the second call, then
+    /// release. Requires `git` on PATH; marked `#[ignore]` so CI
+    /// without git-installed runners skips it silently.
+    #[tokio::test]
+    #[ignore]
+    async fn ensure_then_cached_then_release() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path().join("source.git");
+        let repo_root = tmp.path().join("repos");
+
+        // Init a bare-ish source repo with one commit on branch `main`.
+        let seed_dir = tmp.path().join("seed");
+        std::fs::create_dir_all(&seed_dir).unwrap();
+        let git = |args: &[&str], cwd: &std::path::Path| {
+            let out = std::process::Command::new("git")
+                .args(args)
+                .current_dir(cwd)
+                .output()
+                .unwrap();
+            assert!(
+                out.status.success(),
+                "git {:?} failed: {}",
+                args,
+                String::from_utf8_lossy(&out.stderr)
+            );
+        };
+        git(&["init", "-b", "main"], &seed_dir);
+        git(&["config", "user.email", "t@t"], &seed_dir);
+        git(&["config", "user.name", "t"], &seed_dir);
+        std::fs::write(seed_dir.join("README"), "hi").unwrap();
+        git(&["add", "."], &seed_dir);
+        git(&["commit", "-m", "seed"], &seed_dir);
+        git(
+            &["clone", "--bare", seed_dir.to_str().unwrap(), source.to_str().unwrap()],
+            tmp.path(),
+        );
+
+        let req = RepoEnsureRequest {
+            url: format!("file://{}", source.display()),
+            git_ref: "main".into(),
+            workspace: "workspace:test".into(),
+        };
+
+        // First ensure → fresh clone.
+        let r1 = ensure_repo(&repo_root, &req).await.unwrap();
+        assert!(!r1.cached, "first ensure should not be cached");
+        assert!(!r1.head_sha.is_empty());
+        assert!(std::path::Path::new(&r1.path).join(".git").exists());
+
+        // Second ensure → cache hit.
+        let r2 = ensure_repo(&repo_root, &req).await.unwrap();
+        assert!(r2.cached, "second ensure should hit cache");
+        assert_eq!(r1.head_sha, r2.head_sha);
+        assert_eq!(r1.path, r2.path);
+
+        // Release removes it.
+        let rel = release_repo(
+            &repo_root,
+            &RepoReleaseRequest {
+                url: req.url.clone(),
+                git_ref: req.git_ref.clone(),
+                workspace: req.workspace.clone(),
+            },
+        )
+        .await
+        .unwrap();
+        assert!(rel.removed);
+        assert!(!std::path::Path::new(&r1.path).exists());
+
+        // Idempotent release.
+        let rel2 = release_repo(
+            &repo_root,
+            &RepoReleaseRequest {
+                url: req.url,
+                git_ref: req.git_ref,
+                workspace: req.workspace,
+            },
+        )
+        .await
+        .unwrap();
+        assert!(!rel2.removed);
+    }
+
+    #[test]
+    fn sanitize_component_replaces_traversal() {
+        assert_eq!(sanitize_component(".."), "_");
+        assert_eq!(sanitize_component("../etc"), "_etc");
+        assert_eq!(sanitize_component(""), "_");
+        assert_eq!(sanitize_component("a/b\\c:d"), "a_b_c_d");
+    }
+}