fix(release): 修复发布一致性与并发边界

crates/bat-cas-engine/Cargo.toml

@@ -13,6 +13,7 @@ serde.workspace = true
 serde_json.workspace = true
 tracing.workspace = true
 async-trait.workspace = true
+libc = "0.2"
 
 # 文件系统操作
 tokio = { workspace = true, features = ["fs", "io-util"] }

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

@@ -84,6 +84,22 @@ impl SqliteRefCounter {
         )
         .await?;
 
+        let _release_reference_table = Self::execute_query(
+            &self.pool,
+            sqlx::query(
+                r#"
+                CREATE TABLE IF NOT EXISTS cas_release_references (
+                    release_id TEXT NOT NULL,
+                    ordinal INTEGER NOT NULL CHECK(ordinal >= 0),
+                    object_id TEXT NOT NULL,
+                    released INTEGER NOT NULL CHECK(released IN (0, 1)),
+                    PRIMARY KEY(release_id, ordinal)
+                )
+                "#,
+            ),
+        )
+        .await?;
+
         Ok(())
     }
 
@@ -283,6 +299,95 @@ impl SqliteRefCounter {
             .await?;
         Ok(result.rows_affected() > 0)
     }
+
+    /// Atomically releases one durable release ownership record.
+    ///
+    /// The ownership row and the reference decrement are committed in the
+    /// same SQLite transaction. Retrying the same `(release_id, ordinal)` is
+    /// therefore idempotent, while a different release keeps its own row and
+    /// reference count.
+    pub async fn release_reference_once(
+        &self,
+        release_id: &str,
+        ordinal: u64,
+        hash: &Hash,
+    ) -> Result<bool> {
+        let mut transaction = self.pool.begin().await?;
+        let existing: Option<(String, i64)> = sqlx::query_as(
+            r#"
+            SELECT object_id, released
+            FROM cas_release_references
+            WHERE release_id = ?1 AND ordinal = ?2
+            "#,
+        )
+        .bind(release_id)
+        .bind(ordinal as i64)
+        .fetch_optional(&mut *transaction)
+        .await?;
+
+        if let Some((object_id, released)) = existing {
+            if object_id != hash.to_string() {
+                return Err(CasError::Other(anyhow::anyhow!(
+                    "CAS release ownership mismatch: release={} ordinal={} expected={} actual={}",
+                    release_id,
+                    ordinal,
+                    object_id,
+                    hash
+                )));
+            }
+            if released != 0 {
+                transaction.commit().await?;
+                return Ok(false);
+            }
+            return Err(CasError::Other(anyhow::anyhow!(
+                "CAS release ownership record is not in a retryable state: release={} ordinal={}",
+                release_id,
+                ordinal
+            )));
+        }
+
+        let now = Self::now();
+        let updated: Option<i64> = sqlx::query_scalar(
+            r#"
+            UPDATE cas_objects
+            SET ref_count = ref_count - 1,
+                updated_at = ?1,
+                zero_ref_at = CASE WHEN ref_count = 1 THEN ?1 ELSE zero_ref_at END
+            WHERE hash = ?2 AND ref_count > 0
+            RETURNING ref_count
+            "#,
+        )
+        .bind(now)
+        .bind(hash.to_string())
+        .fetch_optional(&mut *transaction)
+        .await?;
+
+        if updated.is_none() {
+            let exists: Option<i64> =
+                sqlx::query_scalar("SELECT ref_count FROM cas_objects WHERE hash = ?1")
+                    .bind(hash.to_string())
+                    .fetch_optional(&mut *transaction)
+                    .await?;
+            if exists.is_some() {
+                return Err(CasError::ReferenceUnderflow(hash.to_string()));
+            }
+            return Err(CasError::ObjectNotFound(hash.to_string()));
+        }
+
+        sqlx::query(
+            r#"
+            INSERT INTO cas_release_references(release_id, ordinal, object_id, released)
+            VALUES(?1, ?2, ?3, 1)
+            "#,
+        )
+        .bind(release_id)
+        .bind(ordinal as i64)
+        .bind(hash.to_string())
+        .execute(&mut *transaction)
+        .await?;
+        transaction.commit().await?;
+        Ok(true)
+    }
 }
 
 #[cfg(test)]

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

@@ -4,7 +4,25 @@ use crate::error::{CasError, Result};
 use crate::hash::{compute_hash, Hash};
 use crate::refcount::SqliteRefCounter;
 use crate::storage::{FileSystemStorage, Storage, StorageStats};
+use std::fs::OpenOptions;
 use std::path::Path;
+use std::path::PathBuf;
+
+struct CasOperationLock {
+    file: std::fs::File,
+}
+
+impl Drop for CasOperationLock {
+    fn drop(&mut self) {
+        #[cfg(unix)]
+        unsafe {
+            libc::flock(
+                std::os::unix::io::AsRawFd::as_raw_fd(&self.file),
+                libc::LOCK_UN,
+            );
+        }
+    }
+}
 
 /// 文件系统 CAS repository。
 ///
@@ -34,8 +52,16 @@ impl FileSystemCasRepository {
         &self.storage
     }
 
+    async fn acquire_operation_lock(&self) -> Result<CasOperationLock> {
+        let path = self.storage.root().join(".cas-operation.lock");
+        tokio::task::spawn_blocking(move || acquire_operation_lock_sync(path))
+            .await
+            .map_err(|error| CasError::Other(anyhow::anyhow!("CAS lock task failed: {error}")))?
+    }
+
     /// 存储对象并增加引用计数。
     pub async fn store(&self, data: &[u8]) -> Result<Hash> {
+        let _lock = self.acquire_operation_lock().await?;
         let hash = compute_hash(data);
         let existed = self.storage.exists(&hash).await?;
         let stored_hash = self.storage.put(data).await?;
@@ -70,17 +96,20 @@ impl FileSystemCasRepository {
 
     /// 读取对象并验证 Hash。
     pub async fn get(&self, hash: &Hash) -> Result<Vec<u8>> {
+        let _lock = self.acquire_operation_lock().await?;
         let data = self.storage.get(hash).await?;
         Ok(data)
     }
 
     /// 检查对象是否存在。
     pub async fn exists(&self, hash: &Hash) -> Result<bool> {
+        let _lock = self.acquire_operation_lock().await?;
         self.storage.exists(hash).await
     }
 
     /// 增加引用计数。
     pub async fn add_reference(&self, hash: &Hash) -> Result<u64> {
+        let _lock = self.acquire_operation_lock().await?;
         if !self.storage.exists(hash).await? {
             return Err(CasError::ObjectNotFound(hash.to_string()));
         }
@@ -94,22 +123,26 @@ impl FileSystemCasRepository {
 
     /// 减少引用计数。
     pub async fn remove_reference(&self, hash: &Hash) -> Result<u64> {
+        let _lock = self.acquire_operation_lock().await?;
         self.ref_counter.remove_reference(hash).await
     }
 
     /// 获取引用计数。
     pub async fn get_reference_count(&self, hash: &Hash) -> Result<u64> {
+        let _lock = self.acquire_operation_lock().await?;
         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
+        let _lock = self.acquire_operation_lock().await?;
+        self.gc_candidates_unlocked().await
     }
 
     /// 删除引用计数为 0 的对象。
     pub async fn gc(&self) -> Result<u64> {
-        let candidates = self.gc_candidates().await?;
+        let _lock = self.acquire_operation_lock().await?;
+        let candidates = self.gc_candidates_unlocked().await?;
         let mut deleted = 0u64;
 
         for hash in candidates {
@@ -130,10 +163,46 @@ impl FileSystemCasRepository {
         Ok(deleted)
     }
 
+    /// Releases one release-owned reference exactly once.
+    pub async fn release_reference_once(
+        &self,
+        release_id: &str,
+        ordinal: u64,
+        hash: &Hash,
+    ) -> Result<bool> {
+        let _lock = self.acquire_operation_lock().await?;
+        self.ref_counter
+            .release_reference_once(release_id, ordinal, hash)
+            .await
+    }
+
     /// 获取存储统计信息。
     pub async fn stats(&self) -> Result<StorageStats> {
+        let _lock = self.acquire_operation_lock().await?;
         self.storage.stats().await
     }
+
+    async fn gc_candidates_unlocked(&self) -> Result<Vec<Hash>> {
+        self.ref_counter.zero_ref_objects().await
+    }
+}
+
+fn acquire_operation_lock_sync(path: PathBuf) -> Result<CasOperationLock> {
+    let file = OpenOptions::new()
+        .create(true)
+        .truncate(false)
+        .read(true)
+        .write(true)
+        .open(path)?;
+    #[cfg(unix)]
+    {
+        let result =
+            unsafe { libc::flock(std::os::unix::io::AsRawFd::as_raw_fd(&file), libc::LOCK_EX) };
+        if result != 0 {
+            return Err(CasError::Io(std::io::Error::last_os_error()));
+        }
+    }
+    Ok(CasOperationLock { file })
 }
 
 #[cfg(test)]
@@ -228,6 +297,53 @@ mod tests {
         assert_eq!(repo.get_reference_count(&hash).await.unwrap(), 1);
     }
 
+    #[tokio::test]
+    async fn cross_repository_gc_and_store_preserve_object_lifetime() {
+        let (temp_dir, repo) = temp_repo().await;
+        let hash = repo.store(b"cross-process lifetime").await.unwrap();
+        assert_eq!(repo.remove_reference(&hash).await.unwrap(), 0);
+
+        let other = FileSystemCasRepository::new(temp_dir.path()).await.unwrap();
+        let (gc_result, store_result) =
+            tokio::join!(repo.gc(), other.store(b"cross-process lifetime"));
+
+        gc_result.unwrap();
+        assert_eq!(store_result.unwrap(), hash);
+        assert_eq!(other.get_reference_count(&hash).await.unwrap(), 1);
+        assert_eq!(other.get(&hash).await.unwrap(), b"cross-process lifetime");
+    }
+
+    #[tokio::test]
+    async fn release_reference_is_idempotent_after_retry() {
+        let (_temp_dir, repo) = temp_repo().await;
+        let hash = repo.store(b"owned").await.unwrap();
+        assert!(repo
+            .release_reference_once("release-a", 0, &hash)
+            .await
+            .unwrap());
+        assert!(!repo
+            .release_reference_once("release-a", 0, &hash)
+            .await
+            .unwrap());
+        assert_eq!(repo.get_reference_count(&hash).await.unwrap(), 0);
+    }
+
+    #[tokio::test]
+    async fn release_reference_ownership_is_scoped_per_release() {
+        let (_temp_dir, repo) = temp_repo().await;
+        let hash = repo.store(b"shared ownership").await.unwrap();
+        assert_eq!(repo.add_reference(&hash).await.unwrap(), 2);
+        assert!(repo
+            .release_reference_once("release-a", 0, &hash)
+            .await
+            .unwrap());
+        assert!(repo
+            .release_reference_once("release-b", 0, &hash)
+            .await
+            .unwrap());
+        assert_eq!(repo.get_reference_count(&hash).await.unwrap(), 0);
+    }
+
     #[tokio::test]
     async fn corrupted_object_is_detected_through_repository() {
         let (_temp_dir, repo) = temp_repo().await;