feat: implement production cas v1

2026-07-22 11:25:15 +08:00 · 2026-06-28 12:39:17 +08:00
parent dd53e3054e
commit b202d7b231
33 changed files with 1082 additions and 447 deletions
@@ -0,0 +1,236 @@
+//! CAS V1 repository，组合对象存储与引用计数元数据。
+
+use crate::error::{CasError, Result};
+use crate::hash::{compute_hash, Hash};
+use crate::refcount::SqliteRefCounter;
+use crate::storage::{FileSystemStorage, Storage, StorageStats};
+use std::path::Path;
+
+/// 文件系统 CAS repository。
+///
+/// 该类型是 CAS V1 的核心实现。对象文件由 `FileSystemStorage` 管理，
+/// 引用计数和对象元数据由 `SqliteRefCounter` 管理。
+#[derive(Debug, Clone)]
+pub struct FileSystemCasRepository {
+    storage: FileSystemStorage,
+    ref_counter: SqliteRefCounter,
+}
+
+impl FileSystemCasRepository {
+    /// 打开或初始化 CAS repository。
+    pub async fn new(root: impl AsRef<Path>) -> Result<Self> {
+        let root = root.as_ref();
+        let storage = FileSystemStorage::new(root).await?;
+        let ref_counter = SqliteRefCounter::new(root.join("metadata.sqlite")).await?;
+
+        Ok(Self {
+            storage,
+            ref_counter,
+        })
+    }
+
+    /// 返回底层文件存储。
+    pub fn storage(&self) -> &FileSystemStorage {
+        &self.storage
+    }
+
+    /// 存储对象并增加引用计数。
+    pub async fn store(&self, data: &[u8]) -> Result<Hash> {
+        let hash = compute_hash(data);
+        let existed = self.storage.exists(&hash).await;
+        let stored_hash = self.storage.put(data).await?;
+
+        if let Err(error) = self
+            .ref_counter
+            .ensure_object(&stored_hash, data.len() as u64)
+            .await
+        {
+            if !existed {
+                let _ = self.storage.delete(&stored_hash).await;
+            }
+            return Err(error);
+        }
+
+        match self.ref_counter.add_reference(&stored_hash).await {
+            Ok(count) => {
+                debug_assert!(count > 0);
+                Ok(stored_hash)
+            }
+            Err(error) => {
+                if !existed {
+                    let _ = self.storage.delete(&stored_hash).await;
+                }
+                Err(error)
+            }
+        }
+    }
+
+    /// 读取对象并验证 Hash。
+    pub async fn get(&self, hash: &Hash) -> Result<Vec<u8>> {
+        self.storage.get(hash).await
+    }
+
+    /// 检查对象是否存在。
+    pub async fn exists(&self, hash: &Hash) -> bool {
+        self.storage.exists(hash).await
+    }
+
+    /// 增加引用计数。
+    pub async fn add_reference(&self, hash: &Hash) -> Result<u64> {
+        if !self.storage.exists(hash).await {
+            return Err(CasError::ObjectNotFound(hash.to_string()));
+        }
+
+        let size = self.storage.size(hash).await?;
+        self.ref_counter.ensure_object(hash, size).await?;
+        self.ref_counter.add_reference(hash).await
+    }
+
+    /// 减少引用计数。
+    pub async fn remove_reference(&self, hash: &Hash) -> Result<u64> {
+        self.ref_counter.remove_reference(hash).await
+    }
+
+    /// 获取引用计数。
+    pub async fn get_reference_count(&self, hash: &Hash) -> Result<u64> {
+        self.ref_counter.get_reference_count(hash).await
+    }
+
+    /// 返回当前 GC 候选对象。
+    pub async fn gc_candidates(&self) -> Result<Vec<Hash>> {
+        self.ref_counter.zero_ref_objects().await
+    }
+
+    /// 删除引用计数为 0 的对象。
+    pub async fn gc(&self) -> Result<u64> {
+        let candidates = self.gc_candidates().await?;
+        let mut deleted = 0u64;
+
+        for hash in candidates {
+            let ref_count = self.ref_counter.get_reference_count(&hash).await?;
+            if ref_count != 0 {
+                continue;
+            }
+
+            match self.storage.delete(&hash).await {
+                Ok(()) => {
+                    deleted += 1;
+                }
+                Err(CasError::ObjectNotFound(_)) => {}
+                Err(error) => return Err(error),
+            }
+
+            self.ref_counter.delete_metadata(&hash).await?;
+        }
+
+        Ok(deleted)
+    }
+
+    /// 获取存储统计信息。
+    pub async fn stats(&self) -> Result<StorageStats> {
+        self.storage.stats().await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn store_increments_reference_count() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap();
+
+        let hash = repo.store(b"same data").await.unwrap();
+        let second = repo.store(b"same data").await.unwrap();
+
+        assert_eq!(hash, second);
+        assert_eq!(repo.get_reference_count(&hash).await.unwrap(), 2);
+        assert_eq!(repo.stats().await.unwrap().object_count, 1);
+    }
+
+    #[tokio::test]
+    async fn gc_deletes_only_zero_reference_objects() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap();
+
+        let keep = repo.store(b"keep").await.unwrap();
+        let delete = repo.store(b"delete").await.unwrap();
+
+        repo.remove_reference(&delete).await.unwrap();
+
+        assert_eq!(repo.gc().await.unwrap(), 1);
+        assert!(repo.exists(&keep).await);
+        assert!(!repo.exists(&delete).await);
+        assert!(matches!(
+            repo.get_reference_count(&delete).await,
+            Err(CasError::ObjectNotFound(_))
+        ));
+    }
+
+    #[tokio::test]
+    async fn concurrent_store_is_safe_and_counts_references() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let repo = Arc::new(FileSystemCasRepository::new(temp_dir.path()).await.unwrap());
+        let mut tasks = Vec::new();
+
+        for _ in 0..32 {
+            let repo = Arc::clone(&repo);
+            tasks.push(tokio::spawn(async move { repo.store(b"shared").await }));
+        }
+
+        let mut hashes = Vec::new();
+        for task in tasks {
+            hashes.push(task.await.unwrap().unwrap());
+        }
+
+        let first = hashes[0];
+        assert!(hashes.iter().all(|hash| *hash == first));
+        assert_eq!(repo.get_reference_count(&first).await.unwrap(), 32);
+        assert_eq!(repo.stats().await.unwrap().object_count, 1);
+        assert_eq!(repo.get(&first).await.unwrap(), b"shared");
+    }
+
+    #[tokio::test]
+    async fn corrupted_object_is_detected_through_repository() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap();
+
+        let hash = repo.store(b"valid").await.unwrap();
+        tokio::fs::write(repo.storage().object_path(&hash), b"invalid")
+            .await
+            .unwrap();
+
+        assert!(matches!(
+            repo.get(&hash).await,
+            Err(CasError::HashMismatch { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn add_reference_requires_existing_object() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap();
+        let missing = compute_hash(b"missing");
+
+        assert!(matches!(
+            repo.add_reference(&missing).await,
+            Err(CasError::ObjectNotFound(_))
+        ));
+    }
+
+    #[tokio::test]
+    async fn remove_reference_does_not_go_below_zero() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap();
+
+        let hash = repo.store(b"underflow").await.unwrap();
+        assert_eq!(repo.remove_reference(&hash).await.unwrap(), 0);
+
+        assert!(matches!(
+            repo.remove_reference(&hash).await,
+            Err(CasError::ReferenceUnderflow(_))
+        ));
+    }
+}