fix(cas): 消除 store/gc 之间的引用计数竞态

store() 原来先 ensure_object（建行 ref_count=0）再 add_reference，两条语句
非事务；即使连接池 max_connections=1，两语句间的 await 也会释放连接，让并发
gc() 在 ref_count=0 窗口删掉刚存的对象文件与元数据（静默丢数据），或使 store
返回 ObjectNotFound。

- 新增 store_reference：单条 UPSERT 原子建行为 ref_count=1 或 +1，对象行不再
  出现 ref_count=0 的可见窗口；store() 与 repository add_reference() 均改用它。
- 新增 delete_zero_ref_metadata：gc 先原子执行 DELETE ... WHERE ref_count=0，
  仅当 rows_affected>0 才删对象文件；被并发递增抢先时跳过，绝不删除仍被引用对象。
  元数据先删、文件后删，最坏只留无元数据的孤儿文件（可覆盖，无数据丢失）。

新增单测：store 后不暴露 ref_count=0（gc_candidates 为空）、候选被重新引用后
gc 跳过、store_reference 原子建行/递增、delete_zero_ref_metadata 守卫。

对应 issue #18 维护清单 1-5。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -127,6 +127,33 @@ impl SqliteRefCounter {
         Ok(())
     }
 
+    /// 原子地注册对象并把引用计数加一。
+    ///
+    /// 新对象直接以 `ref_count = 1` 建行，已存在对象在同一 UPSERT 语句内 `+1`，
+    /// 因此对象行不会在“已注册但尚未引用”的瞬间以 `ref_count = 0` 暴露给并发 GC。
+    pub async fn store_reference(&self, hash: &Hash, size: u64) -> Result<u64> {
+        let now = Self::now();
+        let count: i64 = sqlx::query_scalar(
+            r#"
+            INSERT INTO cas_objects(hash, size, ref_count, created_at, updated_at, zero_ref_at)
+            VALUES(?1, ?2, 1, ?3, ?3, NULL)
+            ON CONFLICT(hash) DO UPDATE SET
+                ref_count = ref_count + 1,
+                size = excluded.size,
+                updated_at = excluded.updated_at,
+                zero_ref_at = NULL
+            RETURNING ref_count
+            "#,
+        )
+        .bind(hash.to_string())
+        .bind(size as i64)
+        .bind(now)
+        .fetch_one(&self.pool)
+        .await?;
+
+        Ok(count as u64)
+    }
+
     /// 增加对象引用计数。
     pub async fn add_reference(&self, hash: &Hash) -> Result<u64> {
         let now = Self::now();
@@ -241,6 +268,19 @@ impl SqliteRefCounter {
             .await?;
         Ok(result.rows_affected() > 0)
     }
+
+    /// 原子地删除引用计数为 0 的对象元数据。
+    ///
+    /// `WHERE ref_count = 0` 是 GC 的原子闸门：若对象被并发 `store_reference` /
+    /// `add_reference` 抢先递增，本语句 `rows_affected = 0` 返回 `false`，调用方据此
+    /// 跳过删除对象文件，绝不删掉仍被引用的对象。
+    pub async fn delete_zero_ref_metadata(&self, hash: &Hash) -> Result<bool> {
+        let result = sqlx::query("DELETE FROM cas_objects WHERE hash = ?1 AND ref_count = 0")
+            .bind(hash.to_string())
+            .execute(&self.pool)
+            .await?;
+        Ok(result.rows_affected() > 0)
+    }
 }
 
 #[cfg(test)]
@@ -292,4 +332,35 @@ mod tests {
         let error = counter.remove_reference(&hash).await.unwrap_err();
         assert!(matches!(error, CasError::ReferenceUnderflow(_)));
     }
+
+    #[tokio::test]
+    async fn store_reference_creates_at_one_and_increments() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let counter = SqliteRefCounter::new(temp_dir.path().join("metadata.sqlite"))
+            .await
+            .unwrap();
+        let hash = compute_hash(b"atomic");
+
+        // 新对象直接建行为 1，不经过 ref_count=0 阶段。
+        assert_eq!(counter.store_reference(&hash, 6).await.unwrap(), 1);
+        assert_eq!(counter.store_reference(&hash, 6).await.unwrap(), 2);
+    }
+
+    #[tokio::test]
+    async fn delete_zero_ref_metadata_guards_referenced_objects() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let counter = SqliteRefCounter::new(temp_dir.path().join("metadata.sqlite"))
+            .await
+            .unwrap();
+        let hash = compute_hash(b"guarded");
+
+        counter.store_reference(&hash, 7).await.unwrap();
+        // ref_count>0：守卫删除返回 false 且元数据保留。
+        assert!(!counter.delete_zero_ref_metadata(&hash).await.unwrap());
+        assert!(counter.metadata(&hash).await.unwrap().is_some());
+        // 归零后可原子删除。
+        counter.remove_reference(&hash).await.unwrap();
+        assert!(counter.delete_zero_ref_metadata(&hash).await.unwrap());
+        assert!(counter.metadata(&hash).await.unwrap().is_none());
+    }
 }

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

@@ -40,19 +40,13 @@ impl FileSystemCasRepository {
         let existed = self.storage.exists(&hash).await;
         let stored_hash = self.storage.put(data).await?;
 
-        if let Err(error) = self
+        // 单条 UPSERT 原子建行并 +1：对象行不会在 store 期间以 ref_count=0
+        // 暴露给并发 gc，消除“已存对象、尚未加引用”的删除窗口。
+        match self
             .ref_counter
-            .ensure_object(&stored_hash, data.len() as u64)
+            .store_reference(&stored_hash, data.len() as u64)
             .await
         {
-            if !existed {
-                self.delete_new_object_after_metadata_failure(&stored_hash)
-                    .await?;
-            }
-            return Err(error);
-        }
-
-        match self.ref_counter.add_reference(&stored_hash).await {
             Ok(count) => {
                 debug_assert!(count > 0);
                 Ok(stored_hash)
@@ -91,9 +85,10 @@ impl FileSystemCasRepository {
             return Err(CasError::ObjectNotFound(hash.to_string()));
         }
 
+        // 同样走原子 UPSERT 递增，避免 ensure_object 与 add_reference 之间
+        // 出现可被并发 gc 删除的 ref_count=0 窗口。
         let size = self.storage.size(hash).await?;
-        self.ref_counter.ensure_object(hash, size).await?;
+        let count = self.ref_counter.store_reference(hash, size).await?;
-        let count = self.ref_counter.add_reference(hash).await?;
         Ok(count)
     }
 
@@ -118,20 +113,18 @@ impl FileSystemCasRepository {
         let mut deleted = 0u64;
 
         for hash in candidates {
-            let ref_count = self.ref_counter.get_reference_count(&hash).await?;
+            // 先原子删除 ref_count=0 的元数据行；若被并发递增抢先，rows_affected=0，
-            if ref_count != 0 {
+            // 跳过，绝不删除仍被引用对象的文件。删元数据成功后再删文件——最坏只留下
+            // 无元数据的孤儿文件（可被后续覆盖，无数据丢失），而非删掉被引用的内容。
+            if !self.ref_counter.delete_zero_ref_metadata(&hash).await? {
                 continue;
             }
 
             match self.storage.delete(&hash).await {
-                Ok(()) => {
+                Ok(()) | Err(CasError::ObjectNotFound(_)) => {}
-                    deleted += 1;
-                }
-                Err(CasError::ObjectNotFound(_)) => {}
                 Err(error) => return Err(error),
             }
-
+            deleted += 1;
-            self.ref_counter.delete_metadata(&hash).await?;
         }
 
         Ok(deleted)
@@ -207,6 +200,34 @@ mod tests {
         assert_eq!(repo.get(&first).await.unwrap(), b"shared");
     }
 
+    #[tokio::test]
+    async fn store_never_exposes_zero_reference_window() {
+        let (_temp_dir, repo) = temp_repo().await;
+
+        let hash = repo.store(b"payload").await.unwrap();
+        // store 结束后引用计数为 1，绝不会成为 gc 候选。
+        assert_eq!(repo.get_reference_count(&hash).await.unwrap(), 1);
+        assert!(repo.gc_candidates().await.unwrap().is_empty());
+        assert_eq!(repo.gc().await.unwrap(), 0);
+        assert!(repo.exists(&hash).await);
+    }
+
+    #[tokio::test]
+    async fn gc_skips_object_reacquired_before_delete() {
+        let (_temp_dir, repo) = temp_repo().await;
+
+        let hash = repo.store(b"reacquired").await.unwrap();
+        assert_eq!(repo.remove_reference(&hash).await.unwrap(), 0);
+        // 归零后成为 gc 候选。
+        assert_eq!(repo.gc_candidates().await.unwrap(), vec![hash]);
+        // 在删除前被重新引用（模拟 store/gc 竞态中的重新获取）。
+        assert_eq!(repo.add_reference(&hash).await.unwrap(), 1);
+        // gc 的原子闸门应跳过它，对象文件保留。
+        assert_eq!(repo.gc().await.unwrap(), 0);
+        assert!(repo.exists(&hash).await);
+        assert_eq!(repo.get_reference_count(&hash).await.unwrap(), 1);
+    }
+
     #[tokio::test]
     async fn corrupted_object_is_detected_through_repository() {
         let (_temp_dir, repo) = temp_repo().await;