Add local CAS cleanup policy

crates/geth-cas/src/lib.rs

@@ -67,6 +67,15 @@ impl LocalCas {
         Ok(self.blob_path(hash)?.exists())
     }
 
+    pub fn remove(&self, hash: &BlobHash) -> Result<bool, CasError> {
+        let path = self.blob_path(hash)?;
+        if !path.exists() {
+            return Ok(false);
+        }
+        std::fs::remove_file(path)?;
+        Ok(true)
+    }
+
     pub fn list(&self) -> Result<Vec<BlobInfo>, CasError> {
         let blobs = self.root.join("blobs");
         if !blobs.exists() {
@@ -154,4 +163,15 @@ mod tests {
         cas.get_to_path(&info.hash, &out).expect("get");
         assert_eq!(std::fs::read(out).expect("read"), b"hello geth");
     }
+
+    #[test]
+    fn cas_remove_deletes_blob_file() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let cas = LocalCas::new(dir.path());
+        let info = cas.add_bytes(b"remove me").expect("add");
+
+        assert!(cas.remove(&info.hash).expect("remove"));
+        assert!(!cas.has(&info.hash).expect("has after remove"));
+        assert!(!cas.remove(&info.hash).expect("remove missing"));
+    }
 }

crates/geth-cli/src/lib.rs

@@ -185,6 +185,10 @@ pub enum CasCommand {
     Unpin {
         hash: String,
     },
+    Cleanup {
+        #[arg(long)]
+        dry_run: bool,
+    },
     List,
 }
 
@@ -400,6 +404,7 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
             CasCommand::Has { hash } => ControlRequest::CasHas { hash: hash.into() },
             CasCommand::Pin { hash } => ControlRequest::CasPin { hash: hash.into() },
             CasCommand::Unpin { hash } => ControlRequest::CasUnpin { hash: hash.into() },
+            CasCommand::Cleanup { dry_run } => ControlRequest::CasCleanup { dry_run },
             CasCommand::List => ControlRequest::CasList,
         },
         Command::Kv { command } => ControlRequest::ModuleStub {
@@ -596,6 +601,21 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
         ControlResponse::CasPinned { hash, pinned } => {
             println!("{hash}: pinned={pinned}");
         }
+        ControlResponse::CasCleanup {
+            removed,
+            retained_pinned,
+            dry_run,
+        } => {
+            let action = if dry_run { "would remove" } else { "removed" };
+            println!("{action}: {}", removed.len());
+            for hash in removed {
+                println!("  {hash}");
+            }
+            println!("retained_pinned: {}", retained_pinned.len());
+            for hash in retained_pinned {
+                println!("  {hash}");
+            }
+        }
         ControlResponse::CasList { blobs } => {
             for blob in blobs {
                 let pin = if blob.pinned { "pinned" } else { "unpinned" };

crates/geth-control/src/lib.rs

@@ -37,6 +37,9 @@ pub enum ControlRequest {
     CasUnpin {
         hash: BlobHash,
     },
+    CasCleanup {
+        dry_run: bool,
+    },
     CasList,
     KeychainInit {
         admin_key_path: Option<PathBuf>,
@@ -124,6 +127,11 @@ pub enum ControlResponse {
         hash: BlobHash,
         pinned: bool,
     },
+    CasCleanup {
+        removed: Vec<BlobHash>,
+        retained_pinned: Vec<BlobHash>,
+        dry_run: bool,
+    },
     CasList {
         blobs: Vec<CasBlob>,
     },

crates/geth-node/src/lib.rs

@@ -243,6 +243,26 @@ pub fn handle_request(
                 pinned: false,
             })
         }
+        ControlRequest::CasCleanup { dry_run } => {
+            let cas = LocalCas::new(node.paths.cas_dir());
+            let mut removed = Vec::new();
+            let mut retained_pinned = Vec::new();
+            for blob in cas.list()? {
+                if store.is_cas_object_pinned(blob.hash.as_str())? {
+                    retained_pinned.push(blob.hash);
+                    continue;
+                }
+                if !dry_run && cas.remove(&blob.hash)? {
+                    store.delete_cas_object(blob.hash.as_str())?;
+                }
+                removed.push(blob.hash);
+            }
+            Ok(ControlResponse::CasCleanup {
+                removed,
+                retained_pinned,
+                dry_run,
+            })
+        }
         ControlRequest::CasList => {
             let cas = LocalCas::new(node.paths.cas_dir());
             let blobs = cas

crates/geth-store/src/lib.rs

@@ -235,6 +235,14 @@ impl Store {
         Ok(())
     }
 
+    pub fn delete_cas_object(&self, hash: &str) -> Result<(), StoreError> {
+        self.conn
+            .execute("DELETE FROM cas_objects WHERE hash = ?1", params![hash])?;
+        self.conn
+            .execute("DELETE FROM cas_pins WHERE hash = ?1", params![hash])?;
+        Ok(())
+    }
+
     pub fn list_cas_objects(&self) -> Result<Vec<CasObject>, StoreError> {
         let mut stmt = self.conn.prepare(
             "SELECT hash, size_bytes, path FROM cas_objects ORDER BY created_at_ms, hash",
@@ -765,4 +773,19 @@ mod tests {
         assert!(!store.is_cas_object_pinned(hash).expect("unpinned again"));
         assert!(store.list_cas_pins().expect("pins").is_empty());
     }
+
+    #[test]
+    fn deleting_cas_object_removes_pin_metadata() {
+        let store = Store::open_memory().expect("open");
+        let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
+        store
+            .record_cas_object(hash, 10, "/tmp/blob")
+            .expect("record object");
+        store.pin_cas_object(hash).expect("pin");
+
+        store.delete_cas_object(hash).expect("delete object");
+
+        assert!(store.list_cas_objects().expect("objects").is_empty());
+        assert!(store.list_cas_pins().expect("pins").is_empty());
+    }
 }

crates/geth/tests/bootstrap.rs

@@ -168,6 +168,90 @@ fn cas_pin_unpin_updates_local_pin_metadata() {
     }
 }
 
+#[test]
+fn cas_cleanup_removes_unpinned_and_retains_pinned_blobs() {
+    let home = tempfile::tempdir().expect("tempdir");
+    let paths = geth_config::GethPaths::from_home(home.path());
+    let node = geth_node::init_node(&paths).expect("init node");
+    let pinned_path = home.path().join("pinned.txt");
+    let unpinned_path = home.path().join("unpinned.txt");
+    std::fs::write(&pinned_path, b"keep me").expect("write pinned");
+    std::fs::write(&unpinned_path, b"remove me").expect("write unpinned");
+
+    let pinned_hash = match geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasAdd { path: pinned_path },
+    )
+    .expect("add pinned")
+    {
+        geth_control::ControlResponse::CasAdded { hash, .. } => hash,
+        other => panic!("unexpected response: {other:?}"),
+    };
+    let unpinned_hash = match geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasAdd {
+            path: unpinned_path,
+        },
+    )
+    .expect("add unpinned")
+    {
+        geth_control::ControlResponse::CasAdded { hash, .. } => hash,
+        other => panic!("unexpected response: {other:?}"),
+    };
+    geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasPin {
+            hash: pinned_hash.clone(),
+        },
+    )
+    .expect("pin blob");
+
+    let dry_run = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasCleanup { dry_run: true },
+    )
+    .expect("dry-run cleanup");
+    match dry_run {
+        geth_control::ControlResponse::CasCleanup {
+            removed,
+            retained_pinned,
+            dry_run,
+        } => {
+            assert!(dry_run);
+            assert_eq!(removed, vec![unpinned_hash.clone()]);
+            assert_eq!(retained_pinned, vec![pinned_hash.clone()]);
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+    assert!(
+        geth_cas::LocalCas::new(paths.cas_dir())
+            .has(&unpinned_hash)
+            .expect("dry-run keeps unpinned")
+    );
+
+    let cleanup = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::CasCleanup { dry_run: false },
+    )
+    .expect("cleanup");
+    match cleanup {
+        geth_control::ControlResponse::CasCleanup {
+            removed,
+            retained_pinned,
+            dry_run,
+        } => {
+            assert!(!dry_run);
+            assert_eq!(removed, vec![unpinned_hash.clone()]);
+            assert_eq!(retained_pinned, vec![pinned_hash.clone()]);
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    let cas = geth_cas::LocalCas::new(paths.cas_dir());
+    assert!(cas.has(&pinned_hash).expect("pinned retained"));
+    assert!(!cas.has(&unpinned_hash).expect("unpinned removed"));
+}
+
 #[test]
 fn auth_explain_distinguishes_discovered_peer_candidates() {
     let home = tempfile::tempdir().expect("tempdir");