From 1c4debbaa775f2060870dea3f3f5f753f7bf0ed9 Mon Sep 17 00:00:00 2001 From: Yuyi-Oak <1722157266@qq.com> Date: Thu, 16 Jul 2026 07:39:07 -0700 Subject: [PATCH] =?UTF-8?q?fix(cas):=20=E6=B6=88=E9=99=A4=20store/gc=20?= =?UTF-8?q?=E4=B9=8B=E9=97=B4=E7=9A=84=E5=BC=95=E7=94=A8=E8=AE=A1=E6=95=B0?= =?UTF-8?q?=E7=AB=9E=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/bat-cas-engine/src/refcount.rs | 71 +++++++++++++++++++++++++ crates/bat-cas-engine/src/repository.rs | 61 ++++++++++++++------- 2 files changed, 112 insertions(+), 20 deletions(-) diff --git a/crates/bat-cas-engine/src/refcount.rs b/crates/bat-cas-engine/src/refcount.rs index 2665dc9..907ca24 100644 --- a/crates/bat-cas-engine/src/refcount.rs +++ b/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 { + 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 { 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 { + 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()); + } } diff --git a/crates/bat-cas-engine/src/repository.rs b/crates/bat-cas-engine/src/repository.rs index fbcf1c5..1c0f9fc 100644 --- a/crates/bat-cas-engine/src/repository.rs +++ b/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.add_reference(hash).await?; + let count = self.ref_counter.store_reference(hash, size).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?; - if ref_count != 0 { + // 先原子删除 ref_count=0 的元数据行；若被并发递增抢先，rows_affected=0， + // 跳过，绝不删除仍被引用对象的文件。删元数据成功后再删文件——最坏只留下 + // 无元数据的孤儿文件（可被后续覆盖，无数据丢失），而非删掉被引用的内容。 + if !self.ref_counter.delete_zero_ref_metadata(&hash).await? { continue; } match self.storage.delete(&hash).await { - Ok(()) => { - deleted += 1; - } - Err(CasError::ObjectNotFound(_)) => {} + Ok(()) | Err(CasError::ObjectNotFound(_)) => {} Err(error) => return Err(error), } - - self.ref_counter.delete_metadata(&hash).await?; + deleted += 1; } 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;