refactor: address quality findings

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

@@ -46,7 +46,8 @@ impl FileSystemCasRepository {
             .await
         {
             if !existed {
-                let _ = self.storage.delete(&stored_hash).await;
+                self.delete_new_object_after_metadata_failure(&stored_hash)
+                    .await?;
             }
             return Err(error);
         }
@@ -58,13 +59,21 @@ impl FileSystemCasRepository {
             }
             Err(error) => {
                 if !existed {
-                    let _ = self.storage.delete(&stored_hash).await;
+                    self.delete_new_object_after_metadata_failure(&stored_hash)
+                        .await?;
                 }
                 Err(error)
             }
         }
     }
 
+    async fn delete_new_object_after_metadata_failure(&self, hash: &Hash) -> Result<()> {
+        match self.storage.delete(hash).await {
+            Ok(()) | Err(CasError::ObjectNotFound(_)) => Ok(()),
+            Err(error) => Err(error),
+        }
+    }
+
     /// 读取对象并验证 Hash。
     pub async fn get(&self, hash: &Hash) -> Result<Vec<u8>> {
         self.storage.get(hash).await

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

@@ -117,6 +117,53 @@ impl FileSystemStorage {
         file.sync_all().await?;
         Ok(())
     }
+
+    async fn child_directories(path: &Path) -> Result<Vec<PathBuf>> {
+        let mut directories = Vec::new();
+        let mut entries = tokio::fs::read_dir(path).await?;
+
+        while let Some(entry) = entries.next_entry().await? {
+            if entry.file_type().await?.is_dir() {
+                directories.push(entry.path());
+            }
+        }
+
+        Ok(directories)
+    }
+
+    async fn hashes_in_directory(path: &Path) -> Result<Vec<Hash>> {
+        let mut hashes = Vec::new();
+        let mut entries = tokio::fs::read_dir(path).await?;
+
+        while let Some(entry) = entries.next_entry().await? {
+            if let Some(hash) = Self::hash_from_file_entry(entry).await? {
+                hashes.push(hash);
+            }
+        }
+
+        Ok(hashes)
+    }
+
+    async fn hash_from_file_entry(entry: tokio::fs::DirEntry) -> Result<Option<Hash>> {
+        if !entry.file_type().await?.is_file() {
+            return Ok(None);
+        }
+
+        Ok(entry
+            .file_name()
+            .to_str()
+            .and_then(|name| name.parse().ok()))
+    }
+
+    async fn object_shard_directories(&self) -> Result<Vec<PathBuf>> {
+        let mut shard_directories = Vec::new();
+
+        for first_level in Self::child_directories(&self.objects_dir).await? {
+            shard_directories.extend(Self::child_directories(&first_level).await?);
+        }
+
+        Ok(shard_directories)
+    }
 }
 
 #[async_trait]
@@ -212,31 +259,9 @@ impl Storage for FileSystemStorage {
 
     async fn list(&self) -> Result<Vec<Hash>> {
         let mut hashes = Vec::new();
-        let mut first_level = tokio::fs::read_dir(&self.objects_dir).await?;
 
-        while let Some(first_entry) = first_level.next_entry().await? {
-            if !first_entry.file_type().await?.is_dir() {
-                continue;
-            }
-
-            let mut second_level = tokio::fs::read_dir(first_entry.path()).await?;
-            while let Some(second_entry) = second_level.next_entry().await? {
-                if !second_entry.file_type().await?.is_dir() {
-                    continue;
-                }
-
-                let mut files = tokio::fs::read_dir(second_entry.path()).await?;
-                while let Some(file_entry) = files.next_entry().await? {
-                    if !file_entry.file_type().await?.is_file() {
-                        continue;
-                    }
-                    if let Some(file_name) = file_entry.file_name().to_str() {
-                        if let Ok(hash) = file_name.parse() {
-                            hashes.push(hash);
-                        }
-                    }
-                }
-            }
+        for shard_directory in self.object_shard_directories().await? {
+            hashes.extend(Self::hashes_in_directory(&shard_directory).await?);
         }
 
         hashes.sort_by_key(|hash: &Hash| hash.to_string());