refactor: reduce quality findings

crates/bat-cas-engine/src/hash.rs

@@ -29,8 +29,9 @@ impl Hash {
 
     /// 从十六进制字符串解析
     pub fn from_hex(s: &str) -> Result<Self, crate::error::CasError> {
-        let bytes =
-            hex::decode(s).map_err(|_| crate::error::CasError::InvalidHash(s.to_string()))?;
+        let decoded =
+            hex::decode(s).map_err(|_| crate::error::CasError::InvalidHash(s.to_string()));
+        let bytes = decoded?;
 
         if bytes.len() != 32 {
             return Err(crate::error::CasError::InvalidHash(s.to_string()));
@@ -113,6 +114,12 @@ pub async fn compute_file_hash(path: &std::path::Path) -> crate::error::Result<H
 mod tests {
     use super::*;
 
+    fn test_hash_and_hex() -> (Hash, String) {
+        let hash = compute_hash(b"Test data");
+        let hex = hash.to_hex();
+        (hash, hex)
+    }
+
     #[test]
     fn test_compute_hash() {
         let data = b"Hello, World!";
@@ -125,9 +132,7 @@ mod tests {
 
     #[test]
     fn test_hash_to_string() {
-        let data = b"Test data";
-        let hash = compute_hash(data);
-        let hex = hash.to_hex();
+        let (_hash, hex) = test_hash_and_hex();
 
         // 应该是 64 个字符的十六进制字符串
         assert_eq!(hex.len(), 64);
@@ -135,9 +140,7 @@ mod tests {
 
     #[test]
     fn test_hash_from_string() {
-        let data = b"Test data";
-        let hash = compute_hash(data);
-        let hex = hash.to_hex();
+        let (hash, hex) = test_hash_and_hex();
 
         let parsed = Hash::from_hex(&hex).unwrap();
         assert_eq!(hash, parsed);

crates/bat-cas-engine/src/refcount.rs

@@ -2,7 +2,7 @@
 
 use crate::error::{CasError, Result};
 use crate::hash::Hash;
-use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
+use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteQueryResult};
 use sqlx::SqlitePool;
 use std::path::Path;
 use std::str::FromStr;
@@ -54,43 +54,50 @@ impl SqliteRefCounter {
     }
 
     async fn init_schema(&self) -> Result<()> {
-        sqlx::query(
-            r#"
-            CREATE TABLE IF NOT EXISTS cas_objects (
-                hash TEXT PRIMARY KEY NOT NULL,
-                size INTEGER NOT NULL CHECK(size >= 0),
-                ref_count INTEGER NOT NULL CHECK(ref_count >= 0),
-                created_at INTEGER NOT NULL,
-                updated_at INTEGER NOT NULL,
-                zero_ref_at INTEGER
-            )
-            "#,
+        let _schema_result = Self::execute_query(
+            &self.pool,
+            sqlx::query(
+                r#"
+                CREATE TABLE IF NOT EXISTS cas_objects (
+                    hash TEXT PRIMARY KEY NOT NULL,
+                    size INTEGER NOT NULL CHECK(size >= 0),
+                    ref_count INTEGER NOT NULL CHECK(ref_count >= 0),
+                    created_at INTEGER NOT NULL,
+                    updated_at INTEGER NOT NULL,
+                    zero_ref_at INTEGER
+                )
+                "#,
+            ),
         )
-        .execute(&self.pool)
         .await?;
 
-        sqlx::query(
-            r#"
-            CREATE INDEX IF NOT EXISTS idx_cas_objects_ref_count
-            ON cas_objects(ref_count)
-            "#,
+        let _index_result = Self::execute_query(
+            &self.pool,
+            sqlx::query(
+                r#"
+                CREATE INDEX IF NOT EXISTS idx_cas_objects_ref_count
+                ON cas_objects(ref_count)
+                "#,
+            ),
         )
-        .execute(&self.pool)
         .await?;
 
         Ok(())
     }
 
-    fn now() -> i64 {
-        SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .map(|duration| duration.as_secs() as i64)
-            .unwrap_or_default()
+    async fn execute_query<'q>(
+        pool: &SqlitePool,
+        query: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>,
+    ) -> Result<SqliteQueryResult> {
+        let result = query.execute(pool).await?;
+        Ok(result)
     }
 
-    /// 注册对象元数据。如果对象已存在，仅更新大小和更新时间。
-    pub async fn ensure_object(&self, hash: &Hash, size: u64) -> Result<()> {
-        let now = Self::now();
+    fn insert_or_update_object_query(
+        hash: &Hash,
+        size: u64,
+        now: i64,
+    ) -> sqlx::query::Query<'_, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'_>> {
         sqlx::query(
             r#"
             INSERT INTO cas_objects(hash, size, ref_count, created_at, updated_at, zero_ref_at)
@@ -103,8 +110,20 @@ impl SqliteRefCounter {
         .bind(hash.to_string())
         .bind(size as i64)
         .bind(now)
-        .execute(&self.pool)
-        .await?;
+    }
+
+    fn now() -> i64 {
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|duration| duration.as_secs() as i64)
+            .unwrap_or_default()
+    }
+
+    /// 注册对象元数据。如果对象已存在，仅更新大小和更新时间。
+    pub async fn ensure_object(&self, hash: &Hash, size: u64) -> Result<()> {
+        let now = Self::now();
+        let query = Self::insert_or_update_object_query(hash, size, now);
+        let _upsert_result = Self::execute_query(&self.pool, query).await?;
         Ok(())
     }
 

crates/bat-cas-engine/src/repository.rs

@@ -76,7 +76,8 @@ impl FileSystemCasRepository {
 
     /// 读取对象并验证 Hash。
     pub async fn get(&self, hash: &Hash) -> Result<Vec<u8>> {
-        self.storage.get(hash).await
+        let data = self.storage.get(hash).await?;
+        Ok(data)
     }
 
     /// 检查对象是否存在。
@@ -92,7 +93,8 @@ impl FileSystemCasRepository {
 
         let size = self.storage.size(hash).await?;
         self.ref_counter.ensure_object(hash, size).await?;
-        self.ref_counter.add_reference(hash).await
+        let count = self.ref_counter.add_reference(hash).await?;
+        Ok(count)
     }
 
     /// 减少引用计数。
@@ -146,10 +148,15 @@ mod tests {
     use super::*;
     use std::sync::Arc;
 
-    #[tokio::test]
-    async fn store_increments_reference_count() {
+    async fn temp_repo() -> (tempfile::TempDir, FileSystemCasRepository) {
         let temp_dir = tempfile::tempdir().unwrap();
         let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap();
+        (temp_dir, repo)
+    }
+
+    #[tokio::test]
+    async fn store_increments_reference_count() {
+        let (_temp_dir, repo) = temp_repo().await;
 
         let hash = repo.store(b"same data").await.unwrap();
         let second = repo.store(b"same data").await.unwrap();
@@ -161,8 +168,7 @@ mod tests {
 
     #[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 (_temp_dir, repo) = temp_repo().await;
 
         let keep = repo.store(b"keep").await.unwrap();
         let delete = repo.store(b"delete").await.unwrap();
@@ -180,8 +186,8 @@ mod tests {
 
     #[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 (_temp_dir, repo) = temp_repo().await;
+        let repo = Arc::new(repo);
         let mut tasks = Vec::new();
 
         for _ in 0..32 {
@@ -203,13 +209,11 @@ mod tests {
 
     #[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 (_temp_dir, repo) = temp_repo().await;
 
         let hash = repo.store(b"valid").await.unwrap();
-        tokio::fs::write(repo.storage().object_path(&hash), b"invalid")
-            .await
-            .unwrap();
+        let write_result = tokio::fs::write(repo.storage().object_path(&hash), b"invalid").await;
+        write_result.unwrap();
 
         assert!(matches!(
             repo.get(&hash).await,
@@ -219,8 +223,7 @@ mod tests {
 
     #[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 (_temp_dir, repo) = temp_repo().await;
         let missing = compute_hash(b"missing");
 
         assert!(matches!(
@@ -231,8 +234,7 @@ mod tests {
 
     #[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 (_temp_dir, repo) = temp_repo().await;
 
         let hash = repo.store(b"underflow").await.unwrap();
         assert_eq!(repo.remove_reference(&hash).await.unwrap(), 0);

crates/bat-cas-engine/src/storage.rs

@@ -108,16 +108,33 @@ impl FileSystemStorage {
     }
 
     async fn write_temp_file(&self, path: &Path, data: &[u8]) -> Result<()> {
-        let mut file = tokio::fs::OpenOptions::new()
-            .write(true)
-            .create_new(true)
-            .open(path)
-            .await?;
+        let mut file = tokio::fs::File::create_new(path).await?;
         file.write_all(data).await?;
         file.sync_all().await?;
         Ok(())
     }
 
+    async fn validate_existing_object(&self, hash: &Hash) -> Result<bool> {
+        if !self.exists(hash).await {
+            return Ok(false);
+        }
+
+        let existing = self.get(hash).await;
+        match existing {
+            Ok(_) => Ok(true),
+            Err(CasError::HashMismatch { .. }) => Ok(false),
+            Err(error) => Err(error),
+        }
+    }
+
+    async fn remove_temp_file(path: &Path) -> Result<()> {
+        match tokio::fs::remove_file(path).await {
+            Ok(()) => Ok(()),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
+            Err(error) => Err(CasError::Io(error)),
+        }
+    }
+
     async fn child_directories(path: &Path) -> Result<Vec<PathBuf>> {
         let mut directories = Vec::new();
         let mut entries = tokio::fs::read_dir(path).await?;
@@ -164,6 +181,30 @@ impl FileSystemStorage {
 
         Ok(shard_directories)
     }
+
+    async fn commit_temp_file(&self, tmp_path: &Path, path: &Path, hash: &Hash) -> Result<Hash> {
+        let parent = path
+            .parent()
+            .ok_or_else(|| CasError::Other(anyhow::anyhow!("Object path has no parent")))?;
+
+        let rename_result = tokio::fs::rename(tmp_path, path).await;
+        match rename_result {
+            Ok(()) => {
+                Self::sync_directory(parent)?;
+                Ok(*hash)
+            }
+            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
+                Self::remove_temp_file(tmp_path).await?;
+                let stored_data = self.get(hash).await?;
+                drop(stored_data);
+                Ok(*hash)
+            }
+            Err(error) => {
+                Self::remove_temp_file(tmp_path).await?;
+                Err(CasError::Io(error))
+            }
+        }
+    }
 }
 
 #[async_trait]
@@ -172,12 +213,8 @@ impl Storage for FileSystemStorage {
         let hash = compute_hash(data);
         let path = self.object_path(&hash);
 
-        if self.exists(&hash).await {
-            match self.get(&hash).await {
-                Ok(_) => return Ok(hash),
-                Err(CasError::HashMismatch { .. }) => {}
-                Err(error) => return Err(error),
-            }
+        if self.validate_existing_object(&hash).await? {
+            return Ok(hash);
         }
 
         let parent = path
@@ -188,22 +225,7 @@ impl Storage for FileSystemStorage {
 
         let tmp_path = self.temp_path(&hash);
         self.write_temp_file(&tmp_path, data).await?;
-
-        match tokio::fs::rename(&tmp_path, &path).await {
-            Ok(()) => {
-                Self::sync_directory(parent)?;
-                Ok(hash)
-            }
-            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
-                let _ = tokio::fs::remove_file(&tmp_path).await;
-                self.get(&hash).await?;
-                Ok(hash)
-            }
-            Err(error) => {
-                let _ = tokio::fs::remove_file(&tmp_path).await;
-                Err(CasError::Io(error))
-            }
-        }
+        self.commit_temp_file(&tmp_path, &path, &hash).await
     }
 
     async fn get(&self, hash: &Hash) -> Result<Vec<u8>> {
@@ -288,10 +310,15 @@ impl Storage for FileSystemStorage {
 mod tests {
     use super::*;
 
-    #[tokio::test]
-    async fn put_and_get_roundtrip() {
+    async fn temp_storage() -> (tempfile::TempDir, FileSystemStorage) {
         let temp_dir = tempfile::tempdir().unwrap();
         let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap();
+        (temp_dir, storage)
+    }
+
+    #[tokio::test]
+    async fn put_and_get_roundtrip() {
+        let (_temp_dir, storage) = temp_storage().await;
 
         let data = b"Hello, CAS!";
         let hash = storage.put(data).await.unwrap();
@@ -303,8 +330,7 @@ mod tests {
 
     #[tokio::test]
     async fn put_is_deduplicated() {
-        let temp_dir = tempfile::tempdir().unwrap();
-        let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap();
+        let (_temp_dir, storage) = temp_storage().await;
 
         let data = b"Same content";
         let hash1 = storage.put(data).await.unwrap();
@@ -317,25 +343,23 @@ mod tests {
 
     #[tokio::test]
     async fn exists_and_delete() {
-        let temp_dir = tempfile::tempdir().unwrap();
-        let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap();
+        let (_temp_dir, storage) = temp_storage().await;
 
         let hash = storage.put(b"Test data").await.unwrap();
 
         assert!(storage.exists(&hash).await);
-        storage.delete(&hash).await.unwrap();
+        let deleted = storage.delete(&hash).await;
+        deleted.unwrap();
         assert!(!storage.exists(&hash).await);
     }
 
     #[tokio::test]
     async fn corrupted_object_is_rejected() {
-        let temp_dir = tempfile::tempdir().unwrap();
-        let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap();
+        let (_temp_dir, storage) = temp_storage().await;
 
         let hash = storage.put(b"valid data").await.unwrap();
-        tokio::fs::write(storage.object_path(&hash), b"corrupted")
-            .await
-            .unwrap();
+        let write_result = tokio::fs::write(storage.object_path(&hash), b"corrupted").await;
+        write_result.unwrap();
 
         let error = storage.get(&hash).await.unwrap_err();
         assert!(matches!(error, CasError::HashMismatch { .. }));
@@ -345,7 +369,8 @@ mod tests {
     async fn init_reports_path_errors() {
         let temp_dir = tempfile::tempdir().unwrap();
         let file_path = temp_dir.path().join("not-a-directory");
-        tokio::fs::write(&file_path, b"file").await.unwrap();
+        let write_result = tokio::fs::write(&file_path, b"file").await;
+        write_result.unwrap();
 
         let result = FileSystemStorage::new(file_path.join("cas")).await;
         assert!(matches!(result, Err(CasError::Io(_))));