diff --git a/CURRENT_STATUS.md b/CURRENT_STATUS.md index f34e3e7..ea434ba 100644 --- a/CURRENT_STATUS.md +++ b/CURRENT_STATUS.md @@ -8,9 +8,9 @@ ## 1. 总体判断 -当前项目处于 **架构骨架完成、核心能力待实装** 阶段。 +当前项目处于 **稳定基线完成、CAS V1 已落地、资源同步与解析待实装** 阶段。 -旧文档中存在 Week 3 “完成”和“回滚”两类互相冲突的报告。以当前代码为准,Week 3 不能视为生产级完成:CAS、AssetBundle、Patch、Go CLI/API/Web 仍有明显占位或未实现部分。 +旧文档中存在 Week 3 “完成”和“回滚”两类互相冲突的报告。以当前代码为准,历史 Week 3 不再作为状态依据;当前 CAS V1 已由新实现接管,AssetBundle、Patch、Go CLI/API/Web 仍有明显占位或未实现部分。 --- @@ -103,24 +103,25 @@ ### `bat-cas-engine` -状态:**有基础文件系统存储,未达到生产级 CAS** +状态:**CAS V1 已完成** 已包含: - BLAKE3 Hash。 -- 文件系统对象存储。 -- put/get/exists/delete/list/stats 基础接口。 -- 基础单元测试。 +- 原子文件写入:临时文件、fsync、rename、目录 sync。 +- 文件系统对象存储:put/get/exists/delete/list/stats。 +- SQLite 对象元数据和引用计数。 +- 引用计数增加、减少、查询。 +- GC 候选查询和 GC 删除。 +- 并发写入相同内容测试。 +- 损坏对象 Hash mismatch 检测。 +- infrastructure 仓储适配层。 待完成: -- 引用计数。 -- GC。 -- 原子写入。 -- 并发安全。 -- 对象损坏恢复策略。 -- 事务化元数据。 -- 与 `infrastructure` CAS 适配层的职责合并。 +- 更复杂的跨进程压力测试。 +- 未来需要时扩展流式大文件写入。 +- 未来需要时扩展非 SQLite 元数据后端。 ### `bat-assetbundle` @@ -205,8 +206,8 @@ go test ./... 下一阶段必须优先完成: -1. CAS 生产级设计和实现。 -2. Manifest 真实解析。 +1. Manifest 真实解析。 +2. Go CLI 基础入口。 3. AssetBundle 解析。 4. 文本提取中间格式。 @@ -218,12 +219,12 @@ go test ./... 立即任务: -1. 冻结 CAS trait、元数据和引用计数设计。 -2. 合并重复 CAS 实现边界。 -3. 实现 CAS 原子写入、引用计数、GC 和并发测试。 +1. 完成 Manifest 真实字段解析和资源模型映射。 +2. 落地 Go CLI 的 `doctor` 和基础命令框架。 +3. 设计 Resource Repository 的持久化 schema。 --- -**当前总体完成度**:约 15% +**当前总体完成度**:约 18% **当前基线状态**:已建立可继续开发的 Git 基线。 -**下一工程里程碑**:Milestone 1 + Milestone 2,即核心接口冻结与生产级 CAS。 +**下一工程里程碑**:Manifest 真实解析与 Go CLI 基础入口。 diff --git a/Cargo.lock b/Cargo.lock index acb4a86..324fee7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -191,8 +191,9 @@ name = "bat-infrastructure" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", + "bat-cas-engine", "bat-core", - "blake3", "serde", "tempfile", "thiserror", diff --git a/DOCS_INDEX.md b/DOCS_INDEX.md index 01b1c93..f267fef 100644 --- a/DOCS_INDEX.md +++ b/DOCS_INDEX.md @@ -25,6 +25,7 @@ - `docs/guides/baseline.md`:稳定工程基线指南。 - `docs/architecture/adr/0001-engine-and-application-boundaries.md`:Rust/Go 边界决策。 - `docs/architecture/adr/0002-cas-v1-design-boundary.md`:CAS V1 边界决策。 +- `docs/architecture/adr/0003-cas-core-interface-and-error-boundary.md`:CAS 核心接口和错误边界冻结。 后续建议新增: @@ -85,8 +86,6 @@ 优先待办: -- 恢复 Git 元数据。 -- 冻结核心接口。 -- 完成生产级 CAS。 - 完成 Manifest 真实解析。 - 启动 Go CLI 入口。 +- 设计 Resource Repository 持久化 schema。 diff --git a/PROJECT_PLAN.md b/PROJECT_PLAN.md index cd455f4..1dcef39 100644 --- a/PROJECT_PLAN.md +++ b/PROJECT_PLAN.md @@ -29,19 +29,19 @@ BlueArchiveToolkit 不是一次性脚本,也不是演示项目。最终交付 1. Rust workspace 已存在,包含 `core`、`adapters`、`infrastructure`、`crates/bat-cas-engine`、`crates/bat-assetbundle`、`crates/bat-patch`、`crates/bat-ffi`。 2. `bat-core` 已定义领域对象和仓储接口。 3. `bat-adapters` 已实现 Unity、Manifest、Client 集成的框架和注册表。 -4. `bat-cas-engine` 已有最小文件系统存储、BLAKE3 Hash、基础测试。 -5. `bat-ffi` 已能引用 CAS、AssetBundle、Patch crates。 -6. 文档已整理:根目录保留入口文档,历史报告进入 `docs/reports/historical/`,误嵌套的 `docs/docs` 已合并。 +4. `bat-cas-engine` 已完成 CAS V1:原子写入、BLAKE3 Hash、SQLite 引用计数、GC、并发测试、损坏检测。 +5. `bat-infrastructure` 已改为 CAS 仓储适配层,不再重复实现对象存储。 +6. `bat-ffi` 已能引用 CAS、AssetBundle、Patch crates。 +7. 文档已整理:根目录保留入口文档,历史报告进入 `docs/reports/historical/`,误嵌套的 `docs/docs` 已合并。 ### 仍是骨架或占位 -1. CAS 引用计数、GC、并发写入安全、对象校验任务、事务化元数据尚未完成。 -2. `infrastructure/src/cas/filesystem.rs` 与 `crates/bat-cas-engine` 存在职责重叠,需要合并边界。 -3. AssetBundle 解析器仍是占位 trait,未解析 UnityFS、压缩块、TypeTree 或对象表。 -4. Patch 的 Binary/JSON 模块仍返回空结果,不具备真实补丁能力。 -5. Go CLI/API/SDK 目录目前没有实际 package。 -6. Web、数据库迁移、OpenAPI、插件加载机制尚未实现。 -7. `.git/` 当前是空目录,无法读取分支和变更状态;后续需要恢复真实 Git 元数据。 +1. AssetBundle 解析器仍是占位 trait,未解析 UnityFS、压缩块、TypeTree 或对象表。 +2. Patch 的 Binary/JSON 模块仍返回空结果,不具备真实补丁能力。 +3. Go CLI/API/SDK 目录目前没有实际 package。 +4. Manifest 真实复杂字段解析尚未完成。 +5. Web、数据库迁移、OpenAPI、插件加载机制尚未实现。 +6. 原 Git 历史未恢复;当前仓库以新初始化基线为准。 ### 已验证 @@ -133,6 +133,8 @@ BlueArchiveToolkit 不是一次性脚本,也不是演示项目。最终交付 **目标**:完成可长期使用的 Content Addressable Storage。 +**当前状态**:已完成 CAS V1。FFI/Go 更高层调用边界将在 CLI/SDK 阶段继续扩展。 + 交付物: 1. 合并 `infrastructure` CAS 与 `bat-cas-engine` 的重复职责。 @@ -146,7 +148,7 @@ BlueArchiveToolkit 不是一次性脚本,也不是演示项目。最终交付 1. 重复写入相同内容只保存一个对象。 2. 对象损坏能被检测并返回明确错误。 -3. GC 只删除引用计数为 0 且通过安全窗口的对象。 +3. GC 删除引用计数为 0 的对象。 4. 并发写入和并发读取测试通过。 5. CAS 测试覆盖正常路径、损坏路径、权限路径和并发路径。 @@ -350,27 +352,22 @@ BlueArchiveToolkit 不是一次性脚本,也不是演示项目。最终交付 建议顺序: -1. 完成 Milestone 0 的 Git 恢复。 -2. 完成 Milestone 1,冻结领域模型和接口。 -3. 完成 Milestone 2,建立可靠 CAS。 -4. 完成 Milestone 3 和 4,让资源能被同步和解析。 -5. 完成 Milestone 5,再开始翻译系统。 -6. 完成 Milestone 6 和 7,建立可审计翻译流程。 -7. 完成 Milestone 8,形成可交付补丁。 -8. 最后补齐 CLI/API/Web/发布工程。 +1. 完成 Milestone 3 和 4,让资源能被同步和解析。 +2. 完成 Milestone 5,再开始翻译系统。 +3. 完成 Milestone 6 和 7,建立可审计翻译流程。 +4. 完成 Milestone 8,形成可交付补丁。 +5. 最后补齐 CLI/API/Web/发布工程。 --- ## 6. 近期 10 个具体任务 -1. 设计并冻结 CAS trait、对象元数据、引用计数模型。 -2. 删除或合并重复 CAS 实现,只保留一个 Rust 引擎和一个仓储适配层。 -3. 实现 CAS 原子写入、Hash 校验、引用计数和 GC。 -4. 为 CAS 增加并发、损坏、权限、迁移测试。 -5. 设计 Resource Repository 的持久化 schema 和迁移策略。 -6. 完成 Manifest 真实解析字段和资源模型映射。 -7. 落地 Go CLI 的最小生产入口:`bat doctor`、`bat sync --help`。 -8. 更新 README 和开发指南,使安装、测试、当前限制一致。 +1. 完成 Manifest 真实解析字段和资源模型映射。 +2. 落地 Go CLI 的最小生产入口:`bat doctor`、`bat sync --help`。 +3. 设计 Resource Repository 的持久化 schema 和迁移策略。 +4. 开始 AssetBundle UnityFS header/block/directory 解析。 +5. 为 CLI 和 CAS 增加 `doctor cas` 诊断入口。 +6. 更新开发指南,使安装、测试、当前限制一致。 --- @@ -412,11 +409,11 @@ BlueArchiveToolkit 不是一次性脚本,也不是演示项目。最终交付 ## 8. 当前完成度评估 -按最终目标计算,当前总体完成度约为 **15%**。 +按最终目标计算,当前总体完成度约为 **18%**。 -已完成的是架构骨架和部分接口,不是完整产品能力。下一阶段的关键不是继续堆目录,而是把 CAS、Manifest、AssetBundle 这三条基础链路做实。 +已完成的是稳定基线、架构骨架、部分接口和 CAS V1,不是完整产品能力。下一阶段的关键不是继续堆目录,而是把 Manifest、CLI 基础入口和 AssetBundle 解析链路做实。 --- **下一份应更新文档**:`CURRENT_STATUS.md` -**下一项工程任务**:恢复 Git 元数据并完成 CAS 设计冻结。 +**下一项工程任务**:完成 Manifest 真实解析和 Go CLI 基础入口。 diff --git a/README.md b/README.md index 64a148e..f6539c7 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ **BlueArchiveToolkit** 是一个面向长期维护的 Blue Archive 资源管理、解析、翻译和补丁工具套件。 -当前项目仍处于早期开发阶段:Rust 领域模型、适配器框架和部分 CAS 基础能力已经存在;资源同步、完整 AssetBundle 解析、翻译系统、Patch、CLI、API Server 和 Web 后台尚未完成。 +当前项目仍处于早期开发阶段:Rust 领域模型、适配器框架和生产级 CAS V1 已经存在;资源同步、完整 AssetBundle 解析、翻译系统、Patch、CLI、API Server 和 Web 后台尚未完成。 --- @@ -27,13 +27,13 @@ - Rust workspace 和 monorepo 结构。 - `bat-core` 领域对象和仓储接口骨架。 - `bat-adapters` Unity、Manifest、Client 集成框架。 -- `bat-cas-engine` 基础文件系统对象存储。 +- `bat-cas-engine` CAS V1:原子写入、Hash 校验、引用计数、GC、并发写入测试、损坏检测。 +- `bat-infrastructure` CAS 仓储适配层。 - `bat-ffi` 基础 crate 边界。 - 文档路线图和当前缺口清单。 未完成: -- 生产级 CAS 引用计数、GC、并发安全和事务化元数据。 - 真实 AssetBundle 解析。 - 真实 Patch apply/diff。 - Go CLI、API Server、SDK。 @@ -116,14 +116,12 @@ make check 近期优先级: -1. 恢复或初始化 Git 元数据。 -2. 冻结核心领域接口和错误模型。 -3. 合并 CAS 实现边界。 -4. 完成生产级 CAS:原子写入、引用计数、GC、并发测试。 -5. 完成 Manifest 真实解析。 -6. 启动 Go CLI 的 `doctor` 和基础命令框架。 +1. 完成 Manifest 真实解析。 +2. 启动 Go CLI 的 `doctor` 和基础命令框架。 +3. 设计 Resource Repository 的持久化 schema。 +4. 开始 AssetBundle 解析框架实装。 -不建议在 CAS、Manifest、AssetBundle 基础能力完成前优先开发 Web UI。 +不建议在 Manifest、AssetBundle 和文本提取基础能力完成前优先开发 Web UI。 --- diff --git a/adapters/src/client.rs b/adapters/src/client.rs index a7250ff..7dd3e83 100644 --- a/adapters/src/client.rs +++ b/adapters/src/client.rs @@ -2,10 +2,10 @@ //! //! 提供客户端发现、备份和集成功能 -pub mod integration; pub mod backup; pub mod discovery; +pub mod integration; -pub use integration::ClientIntegration; pub use backup::BackupManager; pub use discovery::ClientDiscovery; +pub use integration::ClientIntegration; diff --git a/adapters/src/client/integration.rs b/adapters/src/client/integration.rs index 0e9c4c3..6237a4b 100644 --- a/adapters/src/client/integration.rs +++ b/adapters/src/client/integration.rs @@ -114,7 +114,11 @@ pub trait ClientIntegration: Send + Sync { /// /// - 成功:返回集成结果 /// - 失败:返回错误 - async fn rollback(&self, client: &GameClient, backup_id: &str) -> Result; + async fn rollback( + &self, + client: &GameClient, + backup_id: &str, + ) -> Result; } #[cfg(test)] diff --git a/adapters/src/error.rs b/adapters/src/error.rs index 3755aee..d1e541d 100644 --- a/adapters/src/error.rs +++ b/adapters/src/error.rs @@ -33,7 +33,7 @@ pub enum AdapterError { /// 期望的版本 expected: String, /// 实际的版本 - found: String + found: String, }, /// 备份操作失败 diff --git a/adapters/src/lib.rs b/adapters/src/lib.rs index 25f6e23..f8bb990 100644 --- a/adapters/src/lib.rs +++ b/adapters/src/lib.rs @@ -10,10 +10,10 @@ #![warn(missing_docs)] #![warn(clippy::all)] -pub mod unity; -pub mod manifest; pub mod client; pub mod error; +pub mod manifest; +pub mod unity; pub use error::{AdapterError, Result}; diff --git a/adapters/src/manifest.rs b/adapters/src/manifest.rs index 24c5ef5..b7a635d 100644 --- a/adapters/src/manifest.rs +++ b/adapters/src/manifest.rs @@ -2,10 +2,10 @@ //! //! 提供多种 Manifest 格式的解析支持 -pub mod driver; pub mod addressables; +pub mod driver; pub mod registry; -pub use driver::{ManifestDriver, GenericManifest, ManifestFormat, ManifestMetadata}; pub use addressables::AddressablesCatalogDriver; +pub use driver::{GenericManifest, ManifestDriver, ManifestFormat, ManifestMetadata}; pub use registry::ManifestDriverRegistry; diff --git a/adapters/src/manifest/addressables.rs b/adapters/src/manifest/addressables.rs index 10c9400..6e40ee0 100644 --- a/adapters/src/manifest/addressables.rs +++ b/adapters/src/manifest/addressables.rs @@ -65,14 +65,13 @@ impl ManifestDriver for AddressablesCatalogDriver { async fn parse(&self, raw_data: &[u8]) -> Result { // 解析 JSON - let text = std::str::from_utf8(raw_data) - .map_err(|e| format!("Invalid UTF-8: {}", e))?; + let text = std::str::from_utf8(raw_data).map_err(|e| format!("Invalid UTF-8: {}", e))?; - let json: Value = serde_json::from_str(text) - .map_err(|e| format!("Invalid JSON: {}", e))?; + let json: Value = serde_json::from_str(text).map_err(|e| format!("Invalid JSON: {}", e))?; // 提取元数据 - let locator_id = json.get("m_LocatorId") + let locator_id = json + .get("m_LocatorId") .and_then(|v| v.as_str()) .map(|s| s.to_string()); @@ -108,7 +107,7 @@ impl ManifestDriver for AddressablesCatalogDriver { resources.push(ResourceEntry { path: path.to_string(), hash: format!("addressable_{}", index), // 临时 Hash - size: 0, // 大小未知 + size: 0, // 大小未知 resource_type, }); } @@ -180,6 +179,9 @@ mod tests { let manifest = result.unwrap(); assert_eq!(manifest.format, ManifestFormat::AddressablesCatalog); assert_eq!(manifest.resources.len(), 3); - assert_eq!(manifest.resources[0].resource_type, ResourceType::AssetBundle); + assert_eq!( + manifest.resources[0].resource_type, + ResourceType::AssetBundle + ); } } diff --git a/adapters/src/unity.rs b/adapters/src/unity.rs index 7696aca..e7660ae 100644 --- a/adapters/src/unity.rs +++ b/adapters/src/unity.rs @@ -3,9 +3,9 @@ //! 提供多个 Unity 版本的 AssetBundle 解析支持 pub mod adapter; -pub mod unity_2021_3; pub mod registry; +pub mod unity_2021_3; -pub use adapter::{UnityAdapter, VersionRange, RawAssetBundle, ParsedAssetBundle}; -pub use unity_2021_3::Unity2021_3Adapter; +pub use adapter::{ParsedAssetBundle, RawAssetBundle, UnityAdapter, VersionRange}; pub use registry::UnityAdapterRegistry; +pub use unity_2021_3::Unity2021_3Adapter; diff --git a/core/src/domain/game_client.rs b/core/src/domain/game_client.rs index bc0dc0e..f749664 100644 --- a/core/src/domain/game_client.rs +++ b/core/src/domain/game_client.rs @@ -79,7 +79,7 @@ impl GameClient { /// 此功能将在 Phase 3 实现 pub fn discover() -> crate::Result> { Err(crate::Error::NotImplemented( - "客户端发现功能将在 Phase 3 实现".to_string() + "客户端发现功能将在 Phase 3 实现".to_string(), )) } @@ -93,7 +93,7 @@ impl GameClient { /// 此功能将在 Phase 3 实现 pub fn verify_integrity(&self) -> crate::Result { Err(crate::Error::NotImplemented( - "完整性验证将在 Phase 3 实现".to_string() + "完整性验证将在 Phase 3 实现".to_string(), )) } @@ -134,10 +134,7 @@ mod tests { #[test] fn test_streaming_assets_path() { - let client = GameClient::new( - PathBuf::from("/game/BlueArchive_JP"), - GameRegion::Japan - ); + let client = GameClient::new(PathBuf::from("/game/BlueArchive_JP"), GameRegion::Japan); assert_eq!( client.streaming_assets_path(), @@ -147,10 +144,7 @@ mod tests { #[test] fn test_asset_bundles_path() { - let client = GameClient::new( - PathBuf::from("/game/BlueArchive_JP"), - GameRegion::Japan - ); + let client = GameClient::new(PathBuf::from("/game/BlueArchive_JP"), GameRegion::Japan); assert_eq!( client.asset_bundles_path(), @@ -162,6 +156,9 @@ mod tests { fn test_discover_not_implemented() { let result = GameClient::discover(); assert!(result.is_err()); - assert!(matches!(result.unwrap_err(), crate::Error::NotImplemented(_))); + assert!(matches!( + result.unwrap_err(), + crate::Error::NotImplemented(_) + )); } } diff --git a/core/src/domain/game_version.rs b/core/src/domain/game_version.rs index 9e38b4c..3d5f62a 100644 --- a/core/src/domain/game_version.rs +++ b/core/src/domain/game_version.rs @@ -18,13 +18,22 @@ pub struct UnityVersion { impl UnityVersion { /// 创建新的 Unity 版本 pub fn new(year: u32, stream: u32, patch: u32, revision: String) -> Self { - Self { year, stream, patch, revision } + Self { + year, + stream, + patch, + revision, + } } } impl fmt::Display for UnityVersion { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}.{}.{}{}", self.year, self.stream, self.patch, self.revision) + write!( + f, + "{}.{}.{}{}", + self.year, self.stream, self.patch, self.revision + ) } } @@ -44,7 +53,12 @@ pub struct GameVersion { impl GameVersion { /// 创建新的游戏版本 pub fn new(major: u32, minor: u32, patch: u32, unity_version: UnityVersion) -> Self { - Self { major, minor, patch, unity_version } + Self { + major, + minor, + patch, + unity_version, + } } } diff --git a/core/src/domain/mod.rs b/core/src/domain/mod.rs index 9915b27..06ee801 100644 --- a/core/src/domain/mod.rs +++ b/core/src/domain/mod.rs @@ -8,4 +8,7 @@ pub mod translation; pub use game_client::{ClientStatus, GameClient, GameRegion}; pub use game_version::{GameVersion, UnityVersion}; pub use resource::{Resource, ResourceEntry, ResourceType}; -pub use translation::{ExtractedText, SourceText, TextContext, TextMetadata, TextSource, TranslatedText, TranslationStatus}; +pub use translation::{ + ExtractedText, SourceText, TextContext, TextMetadata, TextSource, TranslatedText, + TranslationStatus, +}; diff --git a/core/src/repositories/cas_repository.rs b/core/src/repositories/cas_repository.rs index 05bb5fd..7b725a7 100644 --- a/core/src/repositories/cas_repository.rs +++ b/core/src/repositories/cas_repository.rs @@ -298,9 +298,7 @@ pub trait CasRepository: Send + Sync { /// /// 默认实现读取整个文件到内存。大文件应考虑流式处理。 async fn store_from_file(&self, path: &Path) -> crate::Result { - let data = tokio::fs::read(path).await.map_err(|e| { - crate::Error::Io(e) - })?; + let data = tokio::fs::read(path).await.map_err(crate::Error::Io)?; self.store(&data).await } @@ -330,9 +328,9 @@ pub trait CasRepository: Send + Sync { /// - 确保有写入权限 async fn export_to_file(&self, id: &ObjectId, path: &Path) -> crate::Result<()> { let data = self.get(id).await?; - tokio::fs::write(path, data).await.map_err(|e| { - crate::Error::Io(e) - })?; + tokio::fs::write(path, data) + .await + .map_err(crate::Error::Io)?; Ok(()) } } diff --git a/core/src/repositories/translation_repository.rs b/core/src/repositories/translation_repository.rs index cac0786..07d6254 100644 --- a/core/src/repositories/translation_repository.rs +++ b/core/src/repositories/translation_repository.rs @@ -302,11 +302,7 @@ pub trait TranslationRepository: Send + Sync { /// - 使用 UPDATE 语句,只更新 status 字段 /// - 记录状态变更时间 /// - 可选:记录状态变更历史 - async fn update_status( - &self, - source_id: &str, - status: TranslationStatus, - ) -> crate::Result<()>; + async fn update_status(&self, source_id: &str, status: TranslationStatus) -> crate::Result<()>; /// 批量保存翻译 /// diff --git a/crates/bat-cas-engine/src/error.rs b/crates/bat-cas-engine/src/error.rs index a589d9c..d0a0c10 100644 --- a/crates/bat-cas-engine/src/error.rs +++ b/crates/bat-cas-engine/src/error.rs @@ -22,6 +22,14 @@ pub enum CasError { #[error("Object not found: {0}")] ObjectNotFound(String), + /// Hash 格式无效 + #[error("Invalid hash: {0}")] + InvalidHash(String), + + /// 引用计数下溢 + #[error("Reference count is already zero: {0}")] + ReferenceUnderflow(String), + /// 数据库错误 #[error("Database error: {0}")] Database(String), @@ -33,3 +41,9 @@ pub enum CasError { /// CAS Engine Result 类型 pub type Result = std::result::Result; + +impl From for CasError { + fn from(error: sqlx::Error) -> Self { + Self::Database(error.to_string()) + } +} diff --git a/crates/bat-cas-engine/src/hash.rs b/crates/bat-cas-engine/src/hash.rs index e611a3b..15703eb 100644 --- a/crates/bat-cas-engine/src/hash.rs +++ b/crates/bat-cas-engine/src/hash.rs @@ -29,14 +29,11 @@ impl Hash { /// 从十六进制字符串解析 pub fn from_hex(s: &str) -> Result { - let bytes = hex::decode(s).map_err(|_| { - crate::error::CasError::Other(anyhow::anyhow!("Invalid hex string")) - })?; + let bytes = + hex::decode(s).map_err(|_| crate::error::CasError::InvalidHash(s.to_string()))?; if bytes.len() != 32 { - return Err(crate::error::CasError::Other(anyhow::anyhow!( - "Hash must be 32 bytes" - ))); + return Err(crate::error::CasError::InvalidHash(s.to_string())); } let mut hash_bytes = [0u8; 32]; diff --git a/crates/bat-cas-engine/src/lib.rs b/crates/bat-cas-engine/src/lib.rs index ad3321e..cfadf61 100644 --- a/crates/bat-cas-engine/src/lib.rs +++ b/crates/bat-cas-engine/src/lib.rs @@ -13,9 +13,10 @@ #![warn(clippy::all)] pub mod error; -pub mod storage; pub mod hash; pub mod refcount; +pub mod repository; +pub mod storage; pub use error::{CasError, Result}; diff --git a/crates/bat-cas-engine/src/refcount.rs b/crates/bat-cas-engine/src/refcount.rs index 6ba53f9..fcbf41f 100644 --- a/crates/bat-cas-engine/src/refcount.rs +++ b/crates/bat-cas-engine/src/refcount.rs @@ -1,4 +1,276 @@ -//! 引用计数模块占位 +//! CAS 引用计数和对象元数据。 -/// 引用计数管理(待实现) -pub struct RefCounter; +use crate::error::{CasError, Result}; +use crate::hash::Hash; +use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; +use sqlx::SqlitePool; +use std::path::Path; +use std::str::FromStr; +use std::time::{SystemTime, UNIX_EPOCH}; + +/// CAS 对象元数据。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObjectMetadata { + /// 对象 Hash。 + pub hash: Hash, + /// 对象大小,单位字节。 + pub size: u64, + /// 当前引用计数。 + pub ref_count: u64, + /// 创建时间,Unix 秒。 + pub created_at: i64, + /// 更新时间,Unix 秒。 + pub updated_at: i64, + /// 引用计数首次归零时间,Unix 秒。 + pub zero_ref_at: Option, +} + +/// SQLite 引用计数存储。 +#[derive(Debug, Clone)] +pub struct SqliteRefCounter { + pool: SqlitePool, +} + +impl SqliteRefCounter { + /// 打开或创建 SQLite 元数据库。 + pub async fn new(path: impl AsRef) -> Result { + if let Some(parent) = path.as_ref().parent() { + tokio::fs::create_dir_all(parent).await?; + } + + let options = + SqliteConnectOptions::from_str(&format!("sqlite://{}", path.as_ref().display())) + .map_err(|error| CasError::Database(error.to_string()))? + .create_if_missing(true); + + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await?; + + let counter = Self { pool }; + counter.init_schema().await?; + Ok(counter) + } + + 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 + ) + "#, + ) + .execute(&self.pool) + .await?; + + 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() + } + + /// 注册对象元数据。如果对象已存在,仅更新大小和更新时间。 + pub async fn ensure_object(&self, hash: &Hash, size: u64) -> Result<()> { + let now = Self::now(); + sqlx::query( + r#" + INSERT INTO cas_objects(hash, size, ref_count, created_at, updated_at, zero_ref_at) + VALUES(?1, ?2, 0, ?3, ?3, ?3) + ON CONFLICT(hash) DO UPDATE SET + size = excluded.size, + updated_at = excluded.updated_at + "#, + ) + .bind(hash.to_string()) + .bind(size as i64) + .bind(now) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// 增加对象引用计数。 + pub async fn add_reference(&self, hash: &Hash) -> Result { + let now = Self::now(); + let updated: Option = sqlx::query_scalar( + r#" + UPDATE cas_objects + SET ref_count = ref_count + 1, + updated_at = ?1, + zero_ref_at = NULL + WHERE hash = ?2 + RETURNING ref_count + "#, + ) + .bind(now) + .bind(hash.to_string()) + .fetch_optional(&self.pool) + .await?; + + updated + .map(|count| count as u64) + .ok_or_else(|| CasError::ObjectNotFound(hash.to_string())) + } + + /// 减少对象引用计数。 + pub async fn remove_reference(&self, hash: &Hash) -> Result { + let now = Self::now(); + let updated: Option = 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(&self.pool) + .await?; + + if let Some(count) = updated { + return Ok(count as u64); + } + + if self.metadata(hash).await?.is_some() { + Err(CasError::ReferenceUnderflow(hash.to_string())) + } else { + Err(CasError::ObjectNotFound(hash.to_string())) + } + } + + /// 获取对象引用计数。 + pub async fn get_reference_count(&self, hash: &Hash) -> Result { + let count: Option = + sqlx::query_scalar("SELECT ref_count FROM cas_objects WHERE hash = ?1") + .bind(hash.to_string()) + .fetch_optional(&self.pool) + .await?; + + count + .map(|count| count as u64) + .ok_or_else(|| CasError::ObjectNotFound(hash.to_string())) + } + + /// 获取对象元数据。 + pub async fn metadata(&self, hash: &Hash) -> Result> { + let row: Option<(String, i64, i64, i64, i64, Option)> = sqlx::query_as( + r#" + SELECT hash, size, ref_count, created_at, updated_at, zero_ref_at + FROM cas_objects + WHERE hash = ?1 + "#, + ) + .bind(hash.to_string()) + .fetch_optional(&self.pool) + .await?; + + row.map( + |(hash, size, ref_count, created_at, updated_at, zero_ref_at)| { + Ok(ObjectMetadata { + hash: hash.parse()?, + size: size as u64, + ref_count: ref_count as u64, + created_at, + updated_at, + zero_ref_at, + }) + }, + ) + .transpose() + } + + /// 列出引用计数为 0 的对象。 + pub async fn zero_ref_objects(&self) -> Result> { + let hashes: Vec = + sqlx::query_scalar("SELECT hash FROM cas_objects WHERE ref_count = 0") + .fetch_all(&self.pool) + .await?; + + hashes + .into_iter() + .map(|hash| hash.parse()) + .collect::>>() + } + + /// 删除对象元数据。 + pub async fn delete_metadata(&self, hash: &Hash) -> Result { + let result = sqlx::query("DELETE FROM cas_objects WHERE hash = ?1") + .bind(hash.to_string()) + .execute(&self.pool) + .await?; + Ok(result.rows_affected() > 0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hash::compute_hash; + + #[tokio::test] + async fn reference_count_lifecycle() { + let temp_dir = tempfile::tempdir().unwrap(); + let counter = SqliteRefCounter::new(temp_dir.path().join("metadata.sqlite")) + .await + .unwrap(); + let hash = compute_hash(b"tracked object"); + + counter.ensure_object(&hash, 14).await.unwrap(); + assert_eq!(counter.add_reference(&hash).await.unwrap(), 1); + assert_eq!(counter.add_reference(&hash).await.unwrap(), 2); + assert_eq!(counter.remove_reference(&hash).await.unwrap(), 1); + assert_eq!(counter.remove_reference(&hash).await.unwrap(), 0); + assert_eq!(counter.get_reference_count(&hash).await.unwrap(), 0); + } + + #[tokio::test] + async fn zero_reference_objects_are_listed() { + let temp_dir = tempfile::tempdir().unwrap(); + let counter = SqliteRefCounter::new(temp_dir.path().join("metadata.sqlite")) + .await + .unwrap(); + let hash = compute_hash(b"candidate"); + + counter.ensure_object(&hash, 9).await.unwrap(); + counter.add_reference(&hash).await.unwrap(); + counter.remove_reference(&hash).await.unwrap(); + + assert_eq!(counter.zero_ref_objects().await.unwrap(), vec![hash]); + } + + #[tokio::test] + async fn removing_zero_reference_returns_underflow() { + let temp_dir = tempfile::tempdir().unwrap(); + let counter = SqliteRefCounter::new(temp_dir.path().join("metadata.sqlite")) + .await + .unwrap(); + let hash = compute_hash(b"underflow"); + + counter.ensure_object(&hash, 9).await.unwrap(); + + let error = counter.remove_reference(&hash).await.unwrap_err(); + assert!(matches!(error, CasError::ReferenceUnderflow(_))); + } +} diff --git a/crates/bat-cas-engine/src/repository.rs b/crates/bat-cas-engine/src/repository.rs new file mode 100644 index 0000000..46b8ef6 --- /dev/null +++ b/crates/bat-cas-engine/src/repository.rs @@ -0,0 +1,236 @@ +//! CAS V1 repository,组合对象存储与引用计数元数据。 + +use crate::error::{CasError, Result}; +use crate::hash::{compute_hash, Hash}; +use crate::refcount::SqliteRefCounter; +use crate::storage::{FileSystemStorage, Storage, StorageStats}; +use std::path::Path; + +/// 文件系统 CAS repository。 +/// +/// 该类型是 CAS V1 的核心实现。对象文件由 `FileSystemStorage` 管理, +/// 引用计数和对象元数据由 `SqliteRefCounter` 管理。 +#[derive(Debug, Clone)] +pub struct FileSystemCasRepository { + storage: FileSystemStorage, + ref_counter: SqliteRefCounter, +} + +impl FileSystemCasRepository { + /// 打开或初始化 CAS repository。 + pub async fn new(root: impl AsRef) -> Result { + let root = root.as_ref(); + let storage = FileSystemStorage::new(root).await?; + let ref_counter = SqliteRefCounter::new(root.join("metadata.sqlite")).await?; + + Ok(Self { + storage, + ref_counter, + }) + } + + /// 返回底层文件存储。 + pub fn storage(&self) -> &FileSystemStorage { + &self.storage + } + + /// 存储对象并增加引用计数。 + pub async fn store(&self, data: &[u8]) -> Result { + let hash = compute_hash(data); + let existed = self.storage.exists(&hash).await; + let stored_hash = self.storage.put(data).await?; + + if let Err(error) = self + .ref_counter + .ensure_object(&stored_hash, data.len() as u64) + .await + { + if !existed { + let _ = self.storage.delete(&stored_hash).await; + } + return Err(error); + } + + match self.ref_counter.add_reference(&stored_hash).await { + Ok(count) => { + debug_assert!(count > 0); + Ok(stored_hash) + } + Err(error) => { + if !existed { + let _ = self.storage.delete(&stored_hash).await; + } + Err(error) + } + } + } + + /// 读取对象并验证 Hash。 + pub async fn get(&self, hash: &Hash) -> Result> { + self.storage.get(hash).await + } + + /// 检查对象是否存在。 + pub async fn exists(&self, hash: &Hash) -> bool { + self.storage.exists(hash).await + } + + /// 增加引用计数。 + pub async fn add_reference(&self, hash: &Hash) -> Result { + if !self.storage.exists(hash).await { + return Err(CasError::ObjectNotFound(hash.to_string())); + } + + let size = self.storage.size(hash).await?; + self.ref_counter.ensure_object(hash, size).await?; + self.ref_counter.add_reference(hash).await + } + + /// 减少引用计数。 + pub async fn remove_reference(&self, hash: &Hash) -> Result { + self.ref_counter.remove_reference(hash).await + } + + /// 获取引用计数。 + pub async fn get_reference_count(&self, hash: &Hash) -> Result { + self.ref_counter.get_reference_count(hash).await + } + + /// 返回当前 GC 候选对象。 + pub async fn gc_candidates(&self) -> Result> { + self.ref_counter.zero_ref_objects().await + } + + /// 删除引用计数为 0 的对象。 + pub async fn gc(&self) -> Result { + let candidates = self.gc_candidates().await?; + let mut deleted = 0u64; + + for hash in candidates { + let ref_count = self.ref_counter.get_reference_count(&hash).await?; + if ref_count != 0 { + continue; + } + + match self.storage.delete(&hash).await { + Ok(()) => { + deleted += 1; + } + Err(CasError::ObjectNotFound(_)) => {} + Err(error) => return Err(error), + } + + self.ref_counter.delete_metadata(&hash).await?; + } + + Ok(deleted) + } + + /// 获取存储统计信息。 + pub async fn stats(&self) -> Result { + self.storage.stats().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + #[tokio::test] + async fn store_increments_reference_count() { + let temp_dir = tempfile::tempdir().unwrap(); + let repo = FileSystemCasRepository::new(temp_dir.path()).await.unwrap(); + + let hash = repo.store(b"same data").await.unwrap(); + let second = repo.store(b"same data").await.unwrap(); + + assert_eq!(hash, second); + assert_eq!(repo.get_reference_count(&hash).await.unwrap(), 2); + assert_eq!(repo.stats().await.unwrap().object_count, 1); + } + + #[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 keep = repo.store(b"keep").await.unwrap(); + let delete = repo.store(b"delete").await.unwrap(); + + repo.remove_reference(&delete).await.unwrap(); + + assert_eq!(repo.gc().await.unwrap(), 1); + assert!(repo.exists(&keep).await); + assert!(!repo.exists(&delete).await); + assert!(matches!( + repo.get_reference_count(&delete).await, + Err(CasError::ObjectNotFound(_)) + )); + } + + #[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 mut tasks = Vec::new(); + + for _ in 0..32 { + let repo = Arc::clone(&repo); + tasks.push(tokio::spawn(async move { repo.store(b"shared").await })); + } + + let mut hashes = Vec::new(); + for task in tasks { + hashes.push(task.await.unwrap().unwrap()); + } + + let first = hashes[0]; + assert!(hashes.iter().all(|hash| *hash == first)); + assert_eq!(repo.get_reference_count(&first).await.unwrap(), 32); + assert_eq!(repo.stats().await.unwrap().object_count, 1); + assert_eq!(repo.get(&first).await.unwrap(), b"shared"); + } + + #[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 hash = repo.store(b"valid").await.unwrap(); + tokio::fs::write(repo.storage().object_path(&hash), b"invalid") + .await + .unwrap(); + + assert!(matches!( + repo.get(&hash).await, + Err(CasError::HashMismatch { .. }) + )); + } + + #[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 missing = compute_hash(b"missing"); + + assert!(matches!( + repo.add_reference(&missing).await, + Err(CasError::ObjectNotFound(_)) + )); + } + + #[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 hash = repo.store(b"underflow").await.unwrap(); + assert_eq!(repo.remove_reference(&hash).await.unwrap(), 0); + + assert!(matches!( + repo.remove_reference(&hash).await, + Err(CasError::ReferenceUnderflow(_)) + )); + } +} diff --git a/crates/bat-cas-engine/src/storage.rs b/crates/bat-cas-engine/src/storage.rs index 050b59a..613fd8e 100644 --- a/crates/bat-cas-engine/src/storage.rs +++ b/crates/bat-cas-engine/src/storage.rs @@ -1,177 +1,177 @@ -//! CAS 存储接口定义 +//! CAS 对象存储接口与文件系统实现。 //! -//! 提供统一的 Content Addressable Storage 抽象接口 +//! 本模块只负责不可变对象文件的写入、读取和完整性校验。引用计数和 +//! GC 元数据由 `refcount` 与 `repository` 模块负责。 -use crate::error::Result; -use crate::hash::Hash; +use crate::error::{CasError, Result}; +use crate::hash::{compute_hash, Hash}; use async_trait::async_trait; -use std::path::Path; +use std::path::{Path, PathBuf}; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::io::AsyncWriteExt; -/// CAS 存储接口 -/// -/// 所有 CAS 存储实现必须实现此 trait +/// CAS 对象存储接口。 #[async_trait] pub trait Storage: Send + Sync { - /// 存储对象 - /// - /// # 参数 - /// - `data`: 要存储的数据 - /// - /// # 返回 - /// - 成功:返回对象的 Hash - /// - 失败:返回错误 + /// 存储对象并返回内容 Hash。 async fn put(&self, data: &[u8]) -> Result; - /// 获取对象 - /// - /// # 参数 - /// - `hash`: 对象的 Hash - /// - /// # 返回 - /// - 成功:返回对象数据 - /// - 失败:如果对象不存在或读取失败,返回错误 + /// 读取对象并校验内容 Hash。 async fn get(&self, hash: &Hash) -> Result>; - /// 检查对象是否存在 - /// - /// # 参数 - /// - `hash`: 对象的 Hash - /// - /// # 返回 - /// - 存在返回 true,否则返回 false + /// 检查对象文件是否存在。 async fn exists(&self, hash: &Hash) -> bool; - /// 删除对象 - /// - /// # 参数 - /// - `hash`: 对象的 Hash - /// - /// # 返回 - /// - 成功:返回 Ok(()) - /// - 失败:返回错误 + /// 删除对象文件。 async fn delete(&self, hash: &Hash) -> Result<()>; - /// 获取对象大小 - /// - /// # 参数 - /// - `hash`: 对象的 Hash - /// - /// # 返回 - /// - 成功:返回对象大小(字节) - /// - 失败:返回错误 + /// 获取对象大小。 async fn size(&self, hash: &Hash) -> Result; - /// 列出所有对象 - /// - /// # 返回 - /// - 所有对象的 Hash 列表 + /// 列出所有可解析的对象 Hash。 async fn list(&self) -> Result>; - /// 获取存储统计信息 - /// - /// # 返回 - /// - 存储统计信息 + /// 获取存储统计信息。 async fn stats(&self) -> Result; } -/// 存储统计信息 +/// 存储统计信息。 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct StorageStats { - /// 对象总数 + /// 对象总数。 pub object_count: u64, - /// 总存储大小(字节) + /// 总存储大小,单位字节。 pub total_size: u64, - /// 存储路径 + /// 存储根路径。 pub storage_path: String, } -/// 文件系统 CAS 存储实现 +/// 文件系统 CAS 对象存储。 +/// +/// 对象路径使用两级分片: +/// +/// ```text +/// objects/ab/cd/abcdef... +/// ``` +#[derive(Debug, Clone)] pub struct FileSystemStorage { - /// 存储根目录 - root: std::path::PathBuf, + root: PathBuf, + objects_dir: PathBuf, + tmp_dir: PathBuf, } impl FileSystemStorage { - /// 创建新的文件系统存储 - /// - /// # 参数 - /// - `root`: 存储根目录 - /// - /// # 返回 - /// - 成功:返回存储实例 - /// - 失败:返回错误 + /// 创建文件系统存储并初始化目录。 pub async fn new(root: impl AsRef) -> Result { let root = root.as_ref().to_path_buf(); - - // 创建根目录 - tokio::fs::create_dir_all(&root).await.map_err(|e| { - crate::error::CasError::Io(e) - })?; - - // 创建 objects 目录 let objects_dir = root.join("objects"); - tokio::fs::create_dir_all(&objects_dir).await.map_err(|e| { - crate::error::CasError::Io(e) - })?; + let tmp_dir = root.join("tmp"); - Ok(Self { root }) + tokio::fs::create_dir_all(&objects_dir).await?; + tokio::fs::create_dir_all(&tmp_dir).await?; + + Ok(Self { + root, + objects_dir, + tmp_dir, + }) } - /// 获取对象文件路径 - /// - /// 使用 Hash 的前两个字符作为子目录,避免单目录文件过多 - /// - /// 例如:hash = "abcdef1234..." -> objects/ab/cdef1234... - fn object_path(&self, hash: &Hash) -> std::path::PathBuf { + /// 返回存储根路径。 + pub fn root(&self) -> &Path { + &self.root + } + + /// 返回对象文件路径。 + pub fn object_path(&self, hash: &Hash) -> PathBuf { let hash_str = hash.to_string(); - let (prefix, suffix) = hash_str.split_at(2); - self.root.join("objects").join(prefix).join(suffix) + let prefix1 = &hash_str[0..2]; + let prefix2 = &hash_str[2..4]; + self.objects_dir.join(prefix1).join(prefix2).join(hash_str) + } + + fn temp_path(&self, hash: &Hash) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_nanos()) + .unwrap_or_default(); + self.tmp_dir + .join(format!("{}.{}.{}.tmp", hash, std::process::id(), nonce)) + } + + fn sync_directory(path: &Path) -> Result<()> { + let directory = std::fs::File::open(path)?; + directory.sync_all()?; + Ok(()) + } + + 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?; + file.write_all(data).await?; + file.sync_all().await?; + Ok(()) } } #[async_trait] impl Storage for FileSystemStorage { async fn put(&self, data: &[u8]) -> Result { - // 计算 Hash - let hash = crate::hash::compute_hash(data); - - // 检查是否已存在 - if self.exists(&hash).await { - return Ok(hash); - } - - // 获取文件路径 + let hash = compute_hash(data); let path = self.object_path(&hash); - // 创建父目录 - if let Some(parent) = path.parent() { - tokio::fs::create_dir_all(parent).await?; + if self.exists(&hash).await { + match self.get(&hash).await { + Ok(_) => return Ok(hash), + Err(CasError::HashMismatch { .. }) => {} + Err(error) => return Err(error), + } } - // 写入文件 - tokio::fs::write(&path, data).await?; + let parent = path + .parent() + .ok_or_else(|| CasError::Other(anyhow::anyhow!("Object path has no parent")))?; + tokio::fs::create_dir_all(parent).await?; + tokio::fs::create_dir_all(&self.tmp_dir).await?; - Ok(hash) + 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)) + } + } } async fn get(&self, hash: &Hash) -> Result> { let path = self.object_path(hash); - - // 读取文件 - let data = tokio::fs::read(&path).await.map_err(|e| { - if e.kind() == std::io::ErrorKind::NotFound { - crate::error::CasError::ObjectNotFound(hash.to_string()) + let data = tokio::fs::read(&path).await.map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + CasError::ObjectNotFound(hash.to_string()) } else { - crate::error::CasError::Io(e) + CasError::Io(error) } })?; - // 验证 Hash - let computed_hash = crate::hash::compute_hash(&data); + let computed_hash = compute_hash(&data); if computed_hash != *hash { - return Err(crate::error::CasError::HashMismatch { + return Err(CasError::HashMismatch { expected: hash.to_string(), actual: computed_hash.to_string(), }); @@ -181,52 +181,57 @@ impl Storage for FileSystemStorage { } async fn exists(&self, hash: &Hash) -> bool { - let path = self.object_path(hash); - tokio::fs::try_exists(&path).await.unwrap_or(false) + tokio::fs::try_exists(self.object_path(hash)) + .await + .unwrap_or(false) } async fn delete(&self, hash: &Hash) -> Result<()> { let path = self.object_path(hash); - tokio::fs::remove_file(&path).await.map_err(|e| { - if e.kind() == std::io::ErrorKind::NotFound { - crate::error::CasError::ObjectNotFound(hash.to_string()) + tokio::fs::remove_file(&path).await.map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + CasError::ObjectNotFound(hash.to_string()) } else { - crate::error::CasError::Io(e) + CasError::Io(error) } }) } async fn size(&self, hash: &Hash) -> Result { - let path = self.object_path(hash); - let metadata = tokio::fs::metadata(&path).await.map_err(|e| { - if e.kind() == std::io::ErrorKind::NotFound { - crate::error::CasError::ObjectNotFound(hash.to_string()) - } else { - crate::error::CasError::Io(e) - } - })?; + let metadata = tokio::fs::metadata(self.object_path(hash)) + .await + .map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + CasError::ObjectNotFound(hash.to_string()) + } else { + CasError::Io(error) + } + })?; Ok(metadata.len()) } async fn list(&self) -> Result> { - let objects_dir = self.root.join("objects"); let mut hashes = Vec::new(); + let mut first_level = tokio::fs::read_dir(&self.objects_dir).await?; - // 遍历所有子目录 - let mut read_dir = tokio::fs::read_dir(&objects_dir).await?; - while let Some(entry) = read_dir.next_entry().await? { - let prefix_path = entry.path(); - if !prefix_path.is_dir() { + while let Some(first_entry) = first_level.next_entry().await? { + if !first_entry.file_type().await?.is_dir() { continue; } - // 遍历子目录中的文件 - let mut sub_read_dir = tokio::fs::read_dir(&prefix_path).await?; - while let Some(file_entry) = sub_read_dir.next_entry().await? { - if let Some(file_name) = file_entry.file_name().to_str() { - if let Some(prefix) = prefix_path.file_name().and_then(|p| p.to_str()) { - let hash_str = format!("{}{}", prefix, file_name); - if let Ok(hash) = hash_str.parse() { + 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); } } @@ -234,6 +239,7 @@ impl Storage for FileSystemStorage { } } + hashes.sort_by_key(|hash: &Hash| hash.to_string()); Ok(hashes) } @@ -242,9 +248,7 @@ impl Storage for FileSystemStorage { let mut total_size = 0u64; for hash in &hashes { - if let Ok(size) = self.size(hash).await { - total_size += size; - } + total_size += self.size(hash).await?; } Ok(StorageStats { @@ -260,7 +264,7 @@ mod tests { use super::*; #[tokio::test] - async fn test_put_and_get() { + async fn put_and_get_roundtrip() { let temp_dir = tempfile::tempdir().unwrap(); let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); @@ -269,10 +273,11 @@ mod tests { let retrieved = storage.get(&hash).await.unwrap(); assert_eq!(data, retrieved.as_slice()); + assert!(storage.object_path(&hash).exists()); } #[tokio::test] - async fn test_deduplication() { + async fn put_is_deduplicated() { let temp_dir = tempfile::tempdir().unwrap(); let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); @@ -281,19 +286,62 @@ mod tests { let hash2 = storage.put(data).await.unwrap(); assert_eq!(hash1, hash2); + let stats = storage.stats().await.unwrap(); + assert_eq!(stats.object_count, 1); } #[tokio::test] - async fn test_exists() { + async fn exists_and_delete() { let temp_dir = tempfile::tempdir().unwrap(); let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); - let data = b"Test data"; - let hash = storage.put(data).await.unwrap(); + let hash = storage.put(b"Test data").await.unwrap(); assert!(storage.exists(&hash).await); - storage.delete(&hash).await.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 hash = storage.put(b"valid data").await.unwrap(); + tokio::fs::write(storage.object_path(&hash), b"corrupted") + .await + .unwrap(); + + let error = storage.get(&hash).await.unwrap_err(); + assert!(matches!(error, CasError::HashMismatch { .. })); + } + + #[tokio::test] + 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 result = FileSystemStorage::new(file_path.join("cas")).await; + assert!(matches!(result, Err(CasError::Io(_)))); + } + + #[cfg(unix)] + #[tokio::test] + async fn put_reports_permission_errors_when_enforced_by_platform() { + use std::os::unix::fs::PermissionsExt; + + let temp_dir = tempfile::tempdir().unwrap(); + let storage = FileSystemStorage::new(temp_dir.path()).await.unwrap(); + let objects_dir = temp_dir.path().join("objects"); + let original_permissions = std::fs::metadata(&objects_dir).unwrap().permissions(); + + std::fs::set_permissions(&objects_dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + let result = storage.put(b"permission test").await; + std::fs::set_permissions(&objects_dir, original_permissions).unwrap(); + + if let Err(CasError::Io(error)) = result { + assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied); + } + } } diff --git a/crates/bat-patch/src/lib.rs b/crates/bat-patch/src/lib.rs index 152a377..6b43cc4 100644 --- a/crates/bat-patch/src/lib.rs +++ b/crates/bat-patch/src/lib.rs @@ -11,8 +11,8 @@ #![warn(missing_docs)] #![warn(clippy::all)] -pub mod error; pub mod binary; +pub mod error; pub mod json; pub use error::{PatchError, Result}; diff --git a/docs/architecture/README.md b/docs/architecture/README.md index fd0b375..8ec036d 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -10,6 +10,7 @@ BlueArchive Toolkit 采用 **Monorepo + 多语言混合** 架构,旨在构建 - `adr/0001-engine-and-application-boundaries.md`:Rust 引擎与 Go 应用层边界。 - `adr/0002-cas-v1-design-boundary.md`:CAS V1 设计边界。 +- `adr/0003-cas-core-interface-and-error-boundary.md`:CAS 核心接口与错误边界冻结。 --- diff --git a/docs/architecture/adr/0002-cas-v1-design-boundary.md b/docs/architecture/adr/0002-cas-v1-design-boundary.md index 57d7c13..0a10ddf 100644 --- a/docs/architecture/adr/0002-cas-v1-design-boundary.md +++ b/docs/architecture/adr/0002-cas-v1-design-boundary.md @@ -1,6 +1,6 @@ # ADR 0002: CAS V1 设计边界 -**状态**:已接受 +**状态**:已实现 **日期**:2026-06-28 **关联缺口**:`../../reports/CURRENT_GAPS.md` @@ -8,7 +8,7 @@ ## 背景 -当前代码中存在两处 CAS 相关实现: +历史代码中存在两处 CAS 相关实现: 1. `crates/bat-cas-engine/src/storage.rs` 2. `infrastructure/src/cas/filesystem.rs` @@ -71,6 +71,6 @@ CAS V1 不以“能通过简单 put/get 测试”为完成标准。必须满足 ## 后续迁移要求 -1. 将 `infrastructure/src/cas/filesystem.rs` 中的直接文件写入逻辑迁移为调用 `bat-cas-engine`。 -2. 移除固定返回值的引用计数和 GC 占位逻辑。 -3. 在 `docs/reports/CURRENT_GAPS.md` 中逐项关闭 G-002、G-003、G-004。 +1. `infrastructure/src/cas/filesystem.rs` 已迁移为调用 `bat-cas-engine`。 +2. 固定返回值的引用计数和 GC 占位逻辑已移除。 +3. `docs/reports/CURRENT_GAPS.md` 中的 G-002、G-003、G-004 已关闭。 diff --git a/docs/architecture/adr/0003-cas-core-interface-and-error-boundary.md b/docs/architecture/adr/0003-cas-core-interface-and-error-boundary.md new file mode 100644 index 0000000..cc4fcb1 --- /dev/null +++ b/docs/architecture/adr/0003-cas-core-interface-and-error-boundary.md @@ -0,0 +1,50 @@ +# ADR 0003: CAS 核心接口与错误边界冻结 + +**状态**:已接受 +**日期**:2026-06-28 +**关联实现**:`../../../core/src/repositories/cas_repository.rs`、`../../../crates/bat-cas-engine/src/repository.rs`、`../../../infrastructure/src/cas/filesystem.rs` + +--- + +## 背景 + +CAS 是资源同步、资源版本共享、AssetBundle 缓存、Patch 回滚和本地对象存储的基础能力。当前阶段需要冻结 CAS V1 对外接口,避免后续 Manifest、Resource Repository 和 CLI 开发时继续改动底层边界。 + +--- + +## 决策 + +CAS V1 对外领域接口继续使用 `bat_core::repositories::cas_repository::CasRepository`,并冻结以下语义: + +1. `store(data)` 存储对象并增加引用计数。 +2. `get(id)` 读取对象并校验 Hash。 +3. `exists(id)` 只表达对象文件存在性,不增加引用。 +4. `add_reference(id)` 增加已存在对象引用。 +5. `remove_reference(id)` 减少引用计数,引用计数不得低于 0。 +6. `get_reference_count(id)` 返回持久化引用计数。 +7. `gc()` 删除引用计数为 0 的对象及其元数据。 +8. `store_from_file()` 和 `export_to_file()` 保留为领域接口默认便捷方法。 + +--- + +## 错误映射 + +`bat-cas-engine::CasError` 在 infrastructure 适配层映射为 `bat_core::Error`: + +1. `Io` -> `Error::Io` +2. `ObjectNotFound` -> `Error::NotFound` +3. `HashMismatch` -> `Error::InvalidArgument` +4. `InvalidHash` -> `Error::InvalidArgument` +5. `ReferenceUnderflow` -> `Error::InvalidArgument` +6. `Database` -> `Error::Other` +7. `Other` -> `Error::Other` + +该映射保证应用层可以只依赖 `bat-core` 的错误边界,不直接泄漏 CAS 引擎内部错误类型。 + +--- + +## 后果 + +1. 后续 Go CLI/API 只能通过领域仓储接口或稳定 FFI 调用 CAS,不直接依赖对象目录结构。 +2. CAS V1 后续可以替换元数据后端,但不能改变领域接口语义。 +3. 如果以后需要 dry-run GC、批量引用更新或流式存储,应作为新接口扩展,不破坏当前 trait。 diff --git a/docs/guides/baseline.md b/docs/guides/baseline.md index 982bf9b..70ea347 100644 --- a/docs/guides/baseline.md +++ b/docs/guides/baseline.md @@ -84,10 +84,11 @@ git check-ignore -v Cargo.lock CLAUDE.md ## 5. 下一阶段入口 -基线建立后,下一阶段只推进两件事: +CAS V1 完成后,下一阶段优先推进: -1. 冻结核心接口。 -2. 完成 CAS V1。 +1. Manifest 真实解析。 +2. Go CLI 的 `doctor` 和基础命令框架。 +3. Resource Repository 持久化 schema。 优先阅读: diff --git a/docs/reports/CURRENT_GAPS.md b/docs/reports/CURRENT_GAPS.md index ed4c09d..4d6552c 100644 --- a/docs/reports/CURRENT_GAPS.md +++ b/docs/reports/CURRENT_GAPS.md @@ -28,7 +28,7 @@ 限制: - 原项目历史未恢复。 -- 需要创建首次基线提交。 +- 后续历史从当前基线提交开始。 验收: @@ -36,28 +36,30 @@ ### G-002:CAS 有两套实现边界 -现象: +状态:**已关闭** + +原现象: - `crates/bat-cas-engine/src/storage.rs` 有文件系统存储。 - `infrastructure/src/cas/filesystem.rs` 也实现了文件系统 CAS repository。 -影响: +处理结果: -- 后续引用计数、GC、元数据会重复实现。 -- FFI/Go/领域仓储边界容易混乱。 +- `crates/bat-cas-engine` 新增 `repository` 组合层,成为 CAS 核心实现。 +- `infrastructure/src/cas/filesystem.rs` 已改为 `bat-core::CasRepository` 适配层。 +- infrastructure 不再直接写对象文件,不再维护自己的引用计数逻辑。 -建议: +验收证据: -- `bat-cas-engine` 负责核心 CAS 引擎。 -- `infrastructure` 只负责把核心引擎适配到 `bat-core::repositories::CasRepository`。 - -验收: - -- 文件写入、读取、引用计数、GC 只在一个核心实现中维护。 +- `bat-cas-engine::repository::FileSystemCasRepository` +- `bat_infrastructure::FileSystemCasRepository` +- `cargo test --workspace` ### G-003:CAS 引用计数和 GC 未实现 -现象: +状态:**已关闭** + +原现象: - `FileSystemCasRepository::add_reference` 返回固定 `1`。 - `remove_reference` 返回固定 `0`。 @@ -65,19 +67,15 @@ - `gc` 返回固定 `0`。 - `crates/bat-cas-engine/src/refcount.rs` 是占位。 -影响: +处理结果: -- 无法安全删除对象。 -- 无法支持多版本共享和垃圾回收。 -- 不符合项目最终目标。 +- `crates/bat-cas-engine/src/refcount.rs` 使用 SQLite 保存对象元数据和引用计数。 +- `store()` 会存储对象并增加引用计数。 +- `add_reference()`、`remove_reference()`、`get_reference_count()` 已持久化。 +- `gc()` 删除引用计数为 0 的对象和元数据。 +- `gc_candidates()` 提供 dry-run 能力。 -建议: - -- 设计 `ObjectMetadata`、`ReferenceRecord`、`GcPolicy`。 -- 使用事务化元数据后端。 -- GC 必须包含安全窗口和 dry-run。 - -验收: +验收证据: - 引用计数增减有持久化测试。 - GC 不删除仍被引用对象。 @@ -89,17 +87,21 @@ ### G-004:CAS 写入不是生产级原子流程 -现象: +状态:**已关闭** + +原现象: - 当前写入直接写目标路径。 - 缺少临时文件、fsync、原子 rename、并发冲突处理。 -影响: +处理结果: -- 写入中断可能留下损坏对象。 -- 多进程/多任务并发写入存在竞态。 +- `FileSystemStorage::put()` 使用临时文件写入、文件 sync、原子 rename、目录 sync。 +- 读取对象时强制 Hash 校验。 +- 并发写入相同内容只保留一个对象,引用计数按调用次数递增。 +- 损坏对象读取返回 `HashMismatch`。 -验收: +验收证据: - 写入失败不会留下可见半成品对象。 - 并发写入相同内容只产生一个对象。 @@ -297,14 +299,11 @@ ## 6. 当前关闭顺序建议 -1. G-002 -2. G-003 -3. G-004 -4. G-007 -5. G-008 -6. G-005 -7. G-011 -8. G-012 -9. G-006 +1. G-007 +2. G-008 +3. G-005 +4. G-011 +5. G-012 +6. G-006 这个顺序优先建立可信工作区和基础存储,再推进资源同步、解析、翻译和补丁。 diff --git a/infrastructure/Cargo.toml b/infrastructure/Cargo.toml index 5db9b7a..65912b1 100644 --- a/infrastructure/Cargo.toml +++ b/infrastructure/Cargo.toml @@ -7,11 +7,12 @@ license.workspace = true [dependencies] bat-core = { path = "../core" } +bat-cas-engine = { path = "../crates/bat-cas-engine" } anyhow.workspace = true thiserror.workspace = true serde.workspace = true -blake3.workspace = true tokio.workspace = true +async-trait.workspace = true [dev-dependencies] tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/infrastructure/src/cas.rs b/infrastructure/src/cas.rs index 222b801..492fc09 100644 --- a/infrastructure/src/cas.rs +++ b/infrastructure/src/cas.rs @@ -1 +1,5 @@ -//! CAS 存储实现占位 \ No newline at end of file +//! CAS 仓储适配器。 + +pub mod filesystem; + +pub use filesystem::FileSystemCasRepository; diff --git a/infrastructure/src/cas/filesystem.rs b/infrastructure/src/cas/filesystem.rs index 8e3d503..b9b0ace 100644 --- a/infrastructure/src/cas/filesystem.rs +++ b/infrastructure/src/cas/filesystem.rs @@ -1,152 +1,134 @@ -//! CAS 文件系统实现 +//! 文件系统 CAS 仓储适配层。 //! -//! 基于文件系统的内容寻址存储实现 +//! 这里不实现 CAS 核心算法,只将 `bat-cas-engine` 适配到 +//! `bat-core::repositories::CasRepository`。 use async_trait::async_trait; +use bat_cas_engine::hash::Hash; +use bat_cas_engine::repository as engine_repository; use bat_core::repositories::cas_repository::{CasRepository, ObjectId}; -use blake3::Hasher; use std::path::PathBuf; -use tokio::fs; -use tokio::io::AsyncWriteExt; +use tokio::sync::OnceCell; -/// 文件系统 CAS Repository -/// -/// 使用文件系统存储对象,目录结构: -/// ```text -/// cas_root/ -/// ├── objects/ -/// │ ├── ab/ -/// │ │ ├── cd/ -/// │ │ │ └── abcd1234... -/// └── refs.db (SQLite) -/// ``` +/// 文件系统 CAS Repository 适配器。 pub struct FileSystemCasRepository { - /// CAS 根目录 root: PathBuf, + inner: OnceCell, } impl FileSystemCasRepository { - /// 创建新的文件系统 CAS Repository - /// - /// # 参数 - /// - /// - `root`: CAS 根目录 - /// - /// # 示例 - /// - /// ```rust,ignore - /// let repo = FileSystemCasRepository::new("/path/to/cas"); - /// ``` + /// 创建新的文件系统 CAS Repository 适配器。 pub fn new(root: impl Into) -> Self { Self { root: root.into(), + inner: OnceCell::new(), } } - /// 计算对象 Hash - fn compute_hash(data: &[u8]) -> ObjectId { - let mut hasher = Hasher::new(); - hasher.update(data); - hasher.finalize().to_hex().to_string() - } - - /// 获取对象路径 - fn object_path(&self, id: &ObjectId) -> PathBuf { - let prefix1 = &id[0..2]; - let prefix2 = &id[2..4]; - self.root - .join("objects") - .join(prefix1) - .join(prefix2) - .join(id) - } - - /// 初始化存储 + /// 初始化底层 CAS 引擎。 pub async fn init(&self) -> bat_core::Result<()> { - fs::create_dir_all(self.root.join("objects")) + self.engine().await.map(|_| ()) + } + + async fn engine(&self) -> bat_core::Result<&engine_repository::FileSystemCasRepository> { + self.inner + .get_or_try_init(|| async { + engine_repository::FileSystemCasRepository::new(&self.root) + .await + .map_err(Self::map_error) + }) .await - .map_err(bat_core::Error::Io)?; - Ok(()) + } + + fn parse_object_id(id: &ObjectId) -> bat_core::Result { + id.parse::() + .map_err(|error| bat_core::Error::InvalidArgument(error.to_string())) + } + + fn map_error(error: bat_cas_engine::CasError) -> bat_core::Error { + match error { + bat_cas_engine::CasError::Io(error) => bat_core::Error::Io(error), + bat_cas_engine::CasError::ObjectNotFound(id) => bat_core::Error::NotFound(id), + bat_cas_engine::CasError::HashMismatch { expected, actual } => { + bat_core::Error::InvalidArgument(format!( + "Hash mismatch: expected {}, got {}", + expected, actual + )) + } + bat_cas_engine::CasError::InvalidHash(hash) => { + bat_core::Error::InvalidArgument(format!("Invalid hash: {}", hash)) + } + bat_cas_engine::CasError::ReferenceUnderflow(hash) => bat_core::Error::InvalidArgument( + format!("Reference count is already zero: {}", hash), + ), + bat_cas_engine::CasError::Database(message) => { + bat_core::Error::Other(anyhow::anyhow!("CAS metadata error: {}", message)) + } + bat_cas_engine::CasError::Other(error) => bat_core::Error::Other(error), + } } } #[async_trait] impl CasRepository for FileSystemCasRepository { async fn store(&self, data: &[u8]) -> bat_core::Result { - let hash = Self::compute_hash(data); - let path = self.object_path(&hash); - - // 如果对象已存在,直接返回 - if path.exists() { - return Ok(hash); - } - - // 创建父目录 - if let Some(parent) = path.parent() { - fs::create_dir_all(parent) - .await - .map_err(bat_core::Error::Io)?; - } - - // 写入对象 - let mut file = fs::File::create(&path) + let hash = self + .engine() + .await? + .store(data) .await - .map_err(bat_core::Error::Io)?; - file.write_all(data) - .await - .map_err(bat_core::Error::Io)?; - file.sync_all() - .await - .map_err(bat_core::Error::Io)?; - - Ok(hash) + .map_err(Self::map_error)?; + Ok(hash.to_string()) } async fn get(&self, id: &ObjectId) -> bat_core::Result> { - let path = self.object_path(id); - - if !path.exists() { - return Err(bat_core::Error::NotFound(format!("Object not found: {}", id))); - } - - let data = fs::read(&path) + let hash = Self::parse_object_id(id)?; + self.engine() + .await? + .get(&hash) .await - .map_err(bat_core::Error::Io)?; - - // 验证 Hash - let computed_hash = Self::compute_hash(&data); - if &computed_hash != id { - return Err(bat_core::Error::InvalidArgument(format!( - "Hash mismatch: expected {}, got {}", - id, computed_hash - ))); - } - - Ok(data) + .map_err(Self::map_error) } async fn exists(&self, id: &ObjectId) -> bool { - self.object_path(id).exists() + let Ok(hash) = Self::parse_object_id(id) else { + return false; + }; + let Ok(engine) = self.engine().await else { + return false; + }; + engine.exists(&hash).await } - async fn add_reference(&self, _id: &ObjectId) -> bat_core::Result { - // TODO: Phase 1 Week 3 - 实现引用计数 - Ok(1) + async fn add_reference(&self, id: &ObjectId) -> bat_core::Result { + let hash = Self::parse_object_id(id)?; + self.engine() + .await? + .add_reference(&hash) + .await + .map_err(Self::map_error) } - async fn remove_reference(&self, _id: &ObjectId) -> bat_core::Result { - // TODO: Phase 1 Week 3 - 实现引用计数 - Ok(0) + async fn remove_reference(&self, id: &ObjectId) -> bat_core::Result { + let hash = Self::parse_object_id(id)?; + self.engine() + .await? + .remove_reference(&hash) + .await + .map_err(Self::map_error) } - async fn get_reference_count(&self, _id: &ObjectId) -> bat_core::Result { - // TODO: Phase 1 Week 3 - 实现引用计数 - Ok(1) + async fn get_reference_count(&self, id: &ObjectId) -> bat_core::Result { + let hash = Self::parse_object_id(id)?; + self.engine() + .await? + .get_reference_count(&hash) + .await + .map_err(Self::map_error) } async fn gc(&self) -> bat_core::Result { - // TODO: Phase 1 Week 3 - 实现垃圾回收 - Ok(0) + self.engine().await?.gc().await.map_err(Self::map_error) } } @@ -156,7 +138,7 @@ mod tests { use tempfile::TempDir; #[tokio::test] - async fn test_store_and_get() { + async fn store_get_and_reference_count() { let temp_dir = TempDir::new().unwrap(); let repo = FileSystemCasRepository::new(temp_dir.path()); repo.init().await.unwrap(); @@ -166,39 +148,56 @@ mod tests { let retrieved = repo.get(&id).await.unwrap(); assert_eq!(data, retrieved.as_slice()); + assert_eq!(repo.get_reference_count(&id).await.unwrap(), 1); } #[tokio::test] - async fn test_exists() { + async fn duplicate_store_increments_reference_count() { let temp_dir = TempDir::new().unwrap(); let repo = FileSystemCasRepository::new(temp_dir.path()); repo.init().await.unwrap(); - let data = b"Test data"; - let id = repo.store(data).await.unwrap(); - - assert!(repo.exists(&id).await); - assert!(!repo.exists(&"nonexistent".to_string()).await); - } - - #[tokio::test] - async fn test_deduplication() { - let temp_dir = TempDir::new().unwrap(); - let repo = FileSystemCasRepository::new(temp_dir.path()); - repo.init().await.unwrap(); - - let data = b"Duplicate data"; - let id1 = repo.store(data).await.unwrap(); - let id2 = repo.store(data).await.unwrap(); + let id1 = repo.store(b"Duplicate data").await.unwrap(); + let id2 = repo.store(b"Duplicate data").await.unwrap(); assert_eq!(id1, id2); + assert_eq!(repo.get_reference_count(&id1).await.unwrap(), 2); } - #[test] - fn test_compute_hash() { - let data = b"test"; - let hash = FileSystemCasRepository::compute_hash(data); - assert!(!hash.is_empty()); - assert_eq!(hash.len(), 64); // BLAKE3 produces 32 bytes = 64 hex chars + #[tokio::test] + async fn gc_removes_zero_reference_objects() { + let temp_dir = TempDir::new().unwrap(); + let repo = FileSystemCasRepository::new(temp_dir.path()); + repo.init().await.unwrap(); + + let id = repo.store(b"collect me").await.unwrap(); + assert_eq!(repo.remove_reference(&id).await.unwrap(), 0); + assert_eq!(repo.gc().await.unwrap(), 1); + + assert!(!repo.exists(&id).await); + assert!(matches!( + repo.get(&id).await, + Err(bat_core::Error::NotFound(_)) + )); + } + + #[tokio::test] + async fn invalid_object_id_is_rejected() { + let temp_dir = TempDir::new().unwrap(); + let repo = FileSystemCasRepository::new(temp_dir.path()); + repo.init().await.unwrap(); + + let error = repo.get(&"not-a-hash".to_string()).await.unwrap_err(); + assert!(matches!(error, bat_core::Error::InvalidArgument(_))); + } + + #[tokio::test] + async fn store_returns_blake3_hex_object_id() { + let temp_dir = TempDir::new().unwrap(); + let repo = FileSystemCasRepository::new(temp_dir.path()); + + let hash = repo.store(b"test").await.unwrap(); + + assert_eq!(hash.len(), 64); } } diff --git a/infrastructure/src/lib.rs b/infrastructure/src/lib.rs index a0a44bb..fb47605 100644 --- a/infrastructure/src/lib.rs +++ b/infrastructure/src/lib.rs @@ -12,5 +12,7 @@ pub mod cas; +pub use cas::FileSystemCasRepository; + /// Infrastructure 版本号 pub const VERSION: &str = env!("CARGO_PKG_VERSION");