diff --git a/CURRENT_STATUS.md b/CURRENT_STATUS.md index 65fad7e..22f0826 100644 --- a/CURRENT_STATUS.md +++ b/CURRENT_STATUS.md @@ -50,7 +50,12 @@ TextUnit scope、source history 和 approved review。worker、TM 复用、人 分开报告；current、state 和 identity 存在但文件被截断或手工修改时返回 `localized.degraded`，只读检查不会自动回滚、删除或修复。双 release 的 `release.status/list/distribution/cleanup` 已由 Rust 从既有状态、manifest、文件系统和 -CAS/reference 元数据统一生成，Go 仅 typed 转发。 +CAS/reference 元数据统一生成，Go 仅 typed 转发。CAS repository 的对象文件、引用计数 +和 GC 通过跨进程操作锁协调，release-local CAS 引用以 `(release, ordinal)` ownership +记录幂等释放；localized publish/rollback 通过 output-root 单写者锁和事务日志恢复 +current、version-state、version 目录。localized release 还写入实际 bytes/BLAKE3 的 +`localized-distribution-manifest.json`，`release.distribution` 使用轻量 metadata 选择， +支持按 destination 对单文件重新校验，不在分发热路径执行完整 release audit。 --- diff --git a/Cargo.lock b/Cargo.lock index a69b024..69cd83e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -103,6 +103,7 @@ dependencies = [ "async-trait", "blake3", "hex", + "libc", "serde", "serde_json", "sqlx", diff --git a/api/openapi/bat-api.yaml b/api/openapi/bat-api.yaml index 6f40c78..aad3e54 100644 --- a/api/openapi/bat-api.yaml +++ b/api/openapi/bat-api.yaml @@ -100,6 +100,11 @@ paths: in: query schema: type: string + - name: destination + in: query + description: Optional release-relative path whose localized bytes and BLAKE3 are revalidated. + schema: + type: string - name: offset in: query schema: diff --git a/crates/bat-cas-engine/Cargo.toml b/crates/bat-cas-engine/Cargo.toml index d0dadf2..60b8f52 100644 --- a/crates/bat-cas-engine/Cargo.toml +++ b/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"] } diff --git a/crates/bat-cas-engine/src/refcount.rs b/crates/bat-cas-engine/src/refcount.rs index b302eaf..918d634 100644 --- a/crates/bat-cas-engine/src/refcount.rs +++ b/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 { + 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 = 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 = + 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)] diff --git a/crates/bat-cas-engine/src/repository.rs b/crates/bat-cas-engine/src/repository.rs index ea7c199..5784bec 100644 --- a/crates/bat-cas-engine/src/repository.rs +++ b/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 { + 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 { + 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> { + let _lock = self.acquire_operation_lock().await?; let data = self.storage.get(hash).await?; Ok(data) } /// 检查对象是否存在。 pub async fn exists(&self, hash: &Hash) -> Result { + let _lock = self.acquire_operation_lock().await?; self.storage.exists(hash).await } /// 增加引用计数。 pub async fn add_reference(&self, hash: &Hash) -> Result { + 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 { + let _lock = self.acquire_operation_lock().await?; self.ref_counter.remove_reference(hash).await } /// 获取引用计数。 pub async fn get_reference_count(&self, hash: &Hash) -> Result { + let _lock = self.acquire_operation_lock().await?; self.ref_counter.get_reference_count(hash).await } /// 返回当前 GC 候选对象。 pub async fn gc_candidates(&self) -> Result> { - 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 { - 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 { + let _lock = self.acquire_operation_lock().await?; + self.ref_counter + .release_reference_once(release_id, ordinal, hash) + .await + } + /// 获取存储统计信息。 pub async fn stats(&self) -> Result { + let _lock = self.acquire_operation_lock().await?; self.storage.stats().await } + + async fn gc_candidates_unlocked(&self) -> Result> { + self.ref_counter.zero_ref_objects().await + } +} + +fn acquire_operation_lock_sync(path: PathBuf) -> Result { + 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; diff --git a/docs/architecture/resource-release-layout.md b/docs/architecture/resource-release-layout.md index a043c7e..4e00233 100644 --- a/docs/architecture/resource-release-layout.md +++ b/docs/architecture/resource-release-layout.md @@ -77,8 +77,11 @@ current -> versions/ # 已汉化后才切换；未汉化状态不发布 versions// # 与官方相对路径一致的汉化资源 localized-patch-manifest.json # localized wrapper + generic PatchManifest 审计输入/结果 + localized-distribution-manifest.json # 实际 localized bytes/hash 的轻量分发索引 .staging// # generic/translation patch 未发布写侧 localized-version-state.json # localized current、官方 source release 和 workflow 状态 + .localized-release.lock # 跨进程单写者锁 + .localized-transaction.json # publish/rollback 崩溃恢复日志 ``` 官方资源发布和汉化发布是两个独立状态： @@ -98,8 +101,16 @@ version state、current symlink、release manifest、文件系统和必要的 CA `release.distribution` 的默认 channel 是 `official`。只有当前或显式历史、路径归属安全、 source relation 正确且 manifest/artifact integrity 通过的 release 才能被选择；staging、 -损坏、缺失、symlink/path escape 或未验证历史项不会回退到另一 channel。返回的 -`resource_root` 和 manifest entry 由 Rust 决定，Go 只做 typed forwarding。 +损坏、缺失、symlink/path escape 或未验证历史项不会回退到另一 channel。Rust 使用已发布 +manifest 做轻量选择，HTTP 热路径不重新执行完整 release audit；localized 必须额外满足 +`localized-distribution-manifest.json` 与 source official manifest 的 destination/URL +集合一致，并返回实际 localized bytes/hash。返回的 `resource_root` 和 manifest entry +由 Rust 决定，Go 只做 typed forwarding；传入 `destination` 时 Rust 会重新校验该文件的 +实际 bytes/BLAKE3。 + +localized publish/rollback 先取得 `.localized-release.lock`，并在 output root 下记录 +`.localized-transaction.json`。current、version-state 和 version 目录的切换按日志阶段 +推进；下一次写操作会先恢复或完成未决事务，避免跨进程并发写入和中断后留下半发布状态。 `release.cleanup` 先生成 dry-run 计划和 `plan_id`，执行时重新计算并比对计划。current、 rollback previous、active/in-progress、localized source official、state/manifest/CAS @@ -163,7 +174,8 @@ GET {public-base-url}/prod-clientpatch.bluearchiveyostar.com//... 默认仅服务 **official download manifest 索引内且 Present + size 匹配** 的文件。需要 localized 或历史 release 时，调用 `release.distribution` 选择 Rust 已验证的 `resource_root`，再由 `/v1/distribution` 或带 `channel`/`release_id` 的 CDN path -转发；Go 不在本地判断健康度，也不回退到 official。 +转发；localized CDN 使用 Rust 返回的实际 bytes/hash 生成 ETag，并在显式请求时校验 +实际文件长度；Go 不在本地判断健康度，也不回退到 official。 ### 3.3 launcher 资源引导兼容 diff --git a/docs/reference/rpc-backend-api.md b/docs/reference/rpc-backend-api.md index 948e7c4..55caabb 100644 --- a/docs/reference/rpc-backend-api.md +++ b/docs/reference/rpc-backend-api.md @@ -178,13 +178,15 @@ SQLite `ResourceRepository`，索引不存在时返回 `ok=true` 且 |---|---|---|---| | `release.status` | 已实现 | `null` | official/localized current、source relation、match、历史 release 和 manifest/artifact/distribution integrity 统一视图。 | | `release.list` | 已实现 | `{ "channel": "official" }` 或 `{ "channel": "localized" }`，可省略 | 对应 namespace 的历史 release 摘要，包含 stable ID、created/published、current pointer、`rollback_available`、lifecycle、`stale`/`damaged`/`referenced`/`unknown`、legacy 和诊断。 | -| `release.distribution` | 已实现 | `{ "channel": "official", "release_id": "...", "offset": 0, "limit": 1000 }`，均可省略 | Rust 选择的 verified `resource_root` 和 download manifest entries；默认 channel 为 official，选择失败返回 `available=false`，不跨 channel fallback。 | +| `release.distribution` | 已实现 | `{ "channel": "official", "release_id": "...", "destination": "...", "offset": 0, "limit": 1000 }`，均可省略 | Rust 选择的 verified `resource_root` 和 download manifest entries；`destination` 用于对单个实际文件重新校验 bytes/BLAKE3；localized 使用发布时生成的实际字节 metadata，不复用 official size/hash；默认 channel 为 official，选择失败返回 `available=false`，不跨 channel fallback。 | | `release.cleanup` | 已实现 | dry-run `{ "execute": false }`；执行 `{ "execute": true, "plan_id": "..." }` | cleanup plan、candidate/retain reasons、blocking references 和 removed paths；执行前会重新生成并比对 `plan_id`。 | `release.status`、`release.list` 和 `release.distribution` 只读现有 official/localized state、current、manifest、文件系统和 CAS/reference 元数据，不创建第二套 release 状态。 -localized 只有在 source official、current pointer、manifest identity、source/target -hash/size 以及 UnityFS/ZIP 最终语义校验全部通过时才可分发；默认官方分发行为不变。 +分发选择使用发布后的轻量 manifest 和文件 size/单文件 BLAKE3 校验，不在 HTTP 热路径 +重新执行完整 release audit；localized 还要求 distribution manifest 的 destination/URL +集合与 source official manifest 一致，并使用发布时记录的实际 localized bytes/hash。 +默认官方分发行为不变。 `release.cleanup` 只删除 Rust 能证明是普通目录且未被 current、rollback、staging、 source、state、manifest、CAS 或未知 ownership 引用的历史项，不修改 current，也不承担 rollback 或 repair。 diff --git a/docs/reports/CURRENT_GAPS.md b/docs/reports/CURRENT_GAPS.md index e0fb437..4fbeb80 100644 --- a/docs/reports/CURRENT_GAPS.md +++ b/docs/reports/CURRENT_GAPS.md @@ -125,6 +125,15 @@ artifact integrity，损坏产物返回 degraded/corrupt，不自动回滚或删 rollback 与 cleanup 保持独立；缺少 generic manifest 的旧 localized release 仍可读， 明确标记 `legacy`/`unknown`，不会被自动重写。 +本轮 P1 一致性修复已完成：CAS repository 的 store/get/reference/GC 使用跨进程操作锁， +release-local CAS 引用通过 durable `(release, ordinal)` ownership ledger 幂等释放；官方 +历史复用只对不可变文件使用 hard link，`translation-tasks.sqlite` 及 WAL/SHM 始终独立 +复制；localized output 使用单写者锁和事务日志恢复 publish/rollback，并在发布时写入 +实际 localized bytes/BLAKE3 的 distribution manifest。分发读取只使用轻量发布 metadata， +保留 path ownership、symlink 和文件完整性检查。P2 尚未由本轮处理：ResourceRepository +更完整的查询/权限/损坏恢复、模糊 TM、bat.sock peer credential/perms、FFI 生命周期、 +资源大小/限额与更强的持久化 fsync 语义仍按后续专项推进。 + ### G-012：Translation Memory V1 已实现，扩展能力仍缺失 Rust `bat` 已提供独立项目级 SQLite TM，记录 raw source/hash、完整 context、release/TextUnit/provider/run provenance，区分 candidate/trusted，只有显式 confirm 才能建立 trusted 记录；worker 只自动复用 trusted 的 raw source + 完整 context exact match，并在复用前执行已批准 Glossary 的确定性 QA。Go `bat-api` 已提供鉴权的 summary/query 只读接口和 confirm 转发，但 Go 不持有 TM 状态。仍缺少模糊匹配和更丰富的导入导出历史能力。 diff --git a/infrastructure/src/cas/filesystem.rs b/infrastructure/src/cas/filesystem.rs index 4f34757..627c431 100644 --- a/infrastructure/src/cas/filesystem.rs +++ b/infrastructure/src/cas/filesystem.rs @@ -30,6 +30,21 @@ impl FileSystemCasRepository { self.engine().await.map(|_| ()) } + /// Releases one release-owned CAS reference exactly once. + pub async fn release_reference_once( + &self, + release_id: &str, + ordinal: u64, + id: &ObjectId, + ) -> bat_core::Result { + let hash = Self::parse_object_id(id)?; + self.engine() + .await? + .release_reference_once(release_id, ordinal, &hash) + .await + .map_err(Self::map_error) + } + async fn engine(&self) -> bat_core::Result<&engine_repository::FileSystemCasRepository> { self.inner .get_or_try_init(|| async { diff --git a/infrastructure/src/lib.rs b/infrastructure/src/lib.rs index 0fcd75c..7fd9ecc 100644 --- a/infrastructure/src/lib.rs +++ b/infrastructure/src/lib.rs @@ -57,13 +57,15 @@ pub use localized_patch::{ inspect_localized_release_artifact, inspect_localized_release_artifact_at, mark_localized_manual_proofreading, read_localized_patch_manifest_at, read_localized_version_state, write_localized_version_state, LocalizedArtifactIntegrityReport, - LocalizedFieldPatch, LocalizedPatchConfig, LocalizedPatchFile, LocalizedPatchInput, - LocalizedPatchIntegrity, LocalizedPatchManifest, LocalizedPatchOperation, - LocalizedPatchOperationMetadata, LocalizedPatchReport, LocalizedPatchRollbackInfo, - LocalizedPatchService, LocalizedRollbackReport, LocalizedStringFieldPatch, - LocalizedTextAssetPatch, LocalizedTranslationWorkflowReport, LocalizedVersionState, - LOCALIZED_CURRENT_LINK, LOCALIZED_PATCH_MANIFEST_FILE, LOCALIZED_PATCH_MANIFEST_VERSION, - LOCALIZED_STAGING_DIR, LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING, + LocalizedDistributionEntry, LocalizedDistributionManifest, LocalizedFieldPatch, + LocalizedPatchConfig, LocalizedPatchFile, LocalizedPatchInput, LocalizedPatchIntegrity, + LocalizedPatchManifest, LocalizedPatchOperation, LocalizedPatchOperationMetadata, + LocalizedPatchReport, LocalizedPatchRollbackInfo, LocalizedPatchService, + LocalizedRollbackReport, LocalizedStringFieldPatch, LocalizedTextAssetPatch, + LocalizedTranslationWorkflowReport, LocalizedVersionState, LOCALIZED_CURRENT_LINK, + LOCALIZED_DISTRIBUTION_MANIFEST_FILE, LOCALIZED_PATCH_MANIFEST_FILE, + LOCALIZED_PATCH_MANIFEST_VERSION, LOCALIZED_STAGING_DIR, + LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING, LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING_LABEL, LOCALIZED_VERSIONS_DIR, LOCALIZED_VERSION_STATE_FILE, LOCALIZED_VERSION_STATE_VERSION, }; diff --git a/infrastructure/src/localized_patch.rs b/infrastructure/src/localized_patch.rs index 13ef9dd..2bfa3ac 100644 --- a/infrastructure/src/localized_patch.rs +++ b/infrastructure/src/localized_patch.rs @@ -7,7 +7,9 @@ use bat_assetbundle::{ }; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, BTreeSet}; -use std::fs; +use std::fs::{self, OpenOptions}; +#[cfg(unix)] +use std::os::unix::io::AsRawFd; use std::path::{Path, PathBuf}; use std::process::Command; use std::time::{SystemTime, UNIX_EPOCH}; @@ -28,6 +30,9 @@ pub const LOCALIZED_VERSIONS_DIR: &str = "versions"; pub const LOCALIZED_VERSION_STATE_FILE: &str = "localized-version-state.json"; /// Per-release patch manifest file name. pub const LOCALIZED_PATCH_MANIFEST_FILE: &str = "localized-patch-manifest.json"; +/// Published per-release distribution metadata with localized bytes. +pub const LOCALIZED_DISTRIBUTION_MANIFEST_FILE: &str = "localized-distribution-manifest.json"; +const LOCALIZED_TRANSACTION_FILE: &str = ".localized-transaction.json"; /// Current localized patch manifest schema version. pub const LOCALIZED_PATCH_MANIFEST_VERSION: u32 = 1; /// Current localized version state schema version. @@ -37,6 +42,85 @@ pub const LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING: &str = "manual_proof /// Human label for `LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING`. pub const LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING_LABEL: &str = "人工校对中"; +#[derive(Debug)] +struct LocalizedOutputLock { + file: std::fs::File, +} + +impl Drop for LocalizedOutputLock { + fn drop(&mut self) { + #[cfg(unix)] + unsafe { + libc::flock(self.file.as_raw_fd(), libc::LOCK_UN); + } + } +} + +impl LocalizedOutputLock { + fn acquire(root: &Path) -> anyhow::Result { + ensure_safe_directory_path(root, "汉化输出目录").map_err(anyhow::Error::msg)?; + fs::create_dir_all(root)?; + ensure_safe_directory_path(root, "汉化输出目录").map_err(anyhow::Error::msg)?; + let path = root.join(".localized-release.lock"); + ensure_safe_file_target(root, &path, "汉化 release 锁").map_err(anyhow::Error::msg)?; + let file = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(&path)?; + #[cfg(unix)] + { + let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) }; + if result != 0 { + return Err(anyhow::anyhow!( + "获取汉化 release 锁失败 {}：{}", + path.display(), + std::io::Error::last_os_error() + )); + } + } + Ok(Self { file }) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct LocalizedReleaseTransaction { + version: u32, + operation: String, + phase: String, + release_id: String, + version_path: PathBuf, + staging_path: Option, + previous_current_target: Option, + current_target: Option, + previous_state_bytes: Option>, + new_state: Option, +} + +impl LocalizedReleaseTransaction { + fn publish( + release_id: &str, + version_path: PathBuf, + staging_path: PathBuf, + previous_current_target: Option, + previous_state_bytes: Option>, + ) -> Self { + Self { + version: 1, + operation: "publish".to_string(), + phase: "prepared".to_string(), + release_id: release_id.to_string(), + version_path, + staging_path: Some(staging_path), + previous_current_target, + current_target: Some(Path::new(LOCALIZED_VERSIONS_DIR).join(release_id)), + previous_state_bytes, + new_state: None, + } + } +} + /// One patch operation against a bundle in an official release. #[derive(Debug, Clone, PartialEq, Eq)] pub struct LocalizedTextAssetPatch { @@ -287,6 +371,32 @@ pub struct LocalizedVersionState { pub updated_unix_seconds: u64, } +/// Actual bytes metadata written alongside a published localized release. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LocalizedDistributionEntry { + /// Official URL associated with this release-relative file. + pub url: String, + /// Release-relative destination. + pub destination: String, + /// Actual localized file size. + pub bytes: u64, + /// BLAKE3 of the actual localized file. + pub blake3: String, +} + +/// Cheap, trusted distribution index generated at localized publication time. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LocalizedDistributionManifest { + /// Metadata schema version. + pub version: u32, + /// Official source release identity. + pub official_release_id: String, + /// Localized release identity. + pub localized_release_id: String, + /// Actual metadata for every official manifest entry. + pub entries: Vec, +} + impl LocalizedVersionState { /// Returns the stable translation workflow status, if set. pub fn translation_workflow_status(&self) -> Option<&str> { @@ -617,6 +727,9 @@ impl LocalizedPatchService { /// Copies the official release, applies patches in staging and publishes it. pub fn publish(&self, config: &LocalizedPatchConfig) -> anyhow::Result { + validate_config(config).map_err(anyhow::Error::msg)?; + let _lock = LocalizedOutputLock::acquire(&config.localized_output_root)?; + recover_localized_transaction(&config.localized_output_root)?; let published_release_id = config.published_release_id().to_string(); let staging = config .localized_output_root @@ -627,7 +740,6 @@ impl LocalizedPatchService { .join(LOCALIZED_VERSIONS_DIR) .join(&published_release_id); let current_path = config.localized_output_root.join(LOCALIZED_CURRENT_LINK); - validate_config(config).map_err(anyhow::Error::msg)?; let state_path = config .localized_output_root .join(LOCALIZED_VERSION_STATE_FILE); @@ -637,9 +749,20 @@ impl LocalizedPatchService { if let Some(target) = previous_current_target.as_deref() { validate_previous_current_target(&config.localized_output_root, target)?; } + let transaction = LocalizedReleaseTransaction::publish( + &published_release_id, + version_path.clone(), + staging.clone(), + previous_current_target.clone(), + previous_state_bytes.clone(), + ); + write_localized_transaction(&config.localized_output_root, &transaction)?; let version_existed_before = version_path.exists(); match self.publish_inner(config, previous_current_target.clone()) { - Ok(report) => Ok(report), + Ok(report) => { + remove_localized_transaction(&config.localized_output_root)?; + Ok(report) + } Err(error) => { if let Err(rollback_error) = rollback_failed_publish( &config.localized_output_root, @@ -654,6 +777,7 @@ impl LocalizedPatchService { "{error}; rollback failed: {rollback_error}" )); } + remove_localized_transaction(&config.localized_output_root)?; Err(error) } } @@ -666,6 +790,8 @@ impl LocalizedPatchService { localized_output_root: &Path, expected_release_id: Option<&str>, ) -> anyhow::Result { + let _lock = LocalizedOutputLock::acquire(localized_output_root)?; + recover_localized_transaction(localized_output_root)?; ensure_safe_directory_path(localized_output_root, "汉化输出目录") .map_err(anyhow::Error::msg)?; let versions_root = localized_output_root.join(LOCALIZED_VERSIONS_DIR); @@ -707,6 +833,7 @@ impl LocalizedPatchService { version_path.join(LOCALIZED_PATCH_MANIFEST_FILE).display() ) })?; + verify_localized_release_files(&version_path, &manifest)?; if manifest.localized_release_id != current_release_id { return Err(anyhow::anyhow!( "manifest release={} 与当前状态 release={} 不一致", @@ -763,23 +890,19 @@ impl LocalizedPatchService { ) })?, ); + verify_localized_release_files(&previous_path, restored_manifest.as_ref().unwrap())?; } - restore_current_symlink( - localized_output_root, - ¤t_path, - manifest.rollback.previous_current_target.as_ref(), - )?; - remove_owned_path(&remove_version_path)?; - - let previous_official_release_id = state.official_release_id; - let previous_workflow_status = state.translation_workflow_status; + let previous_state_bytes = read_file_no_symlink(&state_path, "汉化版本状态") + .map_err(anyhow::Error::msg)? + .ok_or_else(|| anyhow::anyhow!("缺少汉化版本状态"))?; let restored_official_release_id = restored_manifest .as_ref() .map(|manifest| manifest.official_release_id.clone()) - .unwrap_or_else(|| previous_official_release_id.clone()); + .unwrap_or_else(|| state.official_release_id.clone()); + let previous_workflow_status = state.translation_workflow_status.clone(); let translation_workflow_status = - if restored_official_release_id == previous_official_release_id { + if restored_official_release_id == state.official_release_id { previous_workflow_status } else { None @@ -796,7 +919,44 @@ impl LocalizedPatchService { translation_workflow_status, updated_unix_seconds: unix_seconds_now(), }; - write_localized_version_state(localized_output_root, &new_state)?; + let transaction = LocalizedReleaseTransaction { + version: 1, + operation: "rollback".to_string(), + phase: "prepared".to_string(), + release_id: current_release_id.clone(), + version_path: remove_version_path.clone(), + staging_path: None, + previous_current_target: Some( + Path::new(LOCALIZED_VERSIONS_DIR).join(¤t_release_id), + ), + current_target: manifest.rollback.previous_current_target.clone(), + previous_state_bytes: Some(previous_state_bytes), + new_state: Some(new_state.clone()), + }; + write_localized_transaction(localized_output_root, &transaction)?; + let mutation_result = (|| -> anyhow::Result<()> { + restore_current_symlink( + localized_output_root, + ¤t_path, + manifest.rollback.previous_current_target.as_deref(), + )?; + update_localized_transaction_phase(localized_output_root, "current_switched")?; + write_localized_version_state_unlocked(localized_output_root, &new_state)?; + update_localized_transaction_phase(localized_output_root, "state_written")?; + remove_owned_path(&remove_version_path)?; + update_localized_transaction_phase(localized_output_root, "version_removed")?; + Ok(()) + })(); + if let Err(error) = mutation_result { + let recovery = recover_localized_transaction(localized_output_root); + return match recovery { + Ok(()) => Err(error), + Err(recovery_error) => Err(anyhow::anyhow!( + "{error}; localized rollback recovery failed: {recovery_error}" + )), + }; + } + remove_localized_transaction(localized_output_root)?; Ok(LocalizedRollbackReport { command: "localized.rollback", @@ -1018,13 +1178,26 @@ impl LocalizedPatchService { &manifest, &config.unzip_command, )?; + if let Some(distribution) = + build_localized_distribution_manifest(&config.official_release_root, &staging, config)? + { + write_file_atomic( + &staging.join(LOCALIZED_DISTRIBUTION_MANIFEST_FILE), + &serde_json::to_vec_pretty(&distribution)?, + STATE_FILE_MODE, + "localized distribution manifest", + ) + .map_err(anyhow::Error::msg)?; + } fs::create_dir_all(config.localized_output_root.join(LOCALIZED_VERSIONS_DIR))?; fs::rename(&staging, &version_path)?; + update_localized_transaction_phase(&config.localized_output_root, "version_published")?; switch_current_symlink( &config.localized_output_root, ¤t_path, config.published_release_id(), )?; + update_localized_transaction_phase(&config.localized_output_root, "current_switched")?; let state = LocalizedVersionState { state_version: LOCALIZED_VERSION_STATE_VERSION, @@ -1041,6 +1214,7 @@ impl LocalizedPatchService { "汉化版本状态", ) .map_err(anyhow::Error::msg)?; + update_localized_transaction_phase(&config.localized_output_root, "state_written")?; let integrity = verify_published_localized_release( &config.official_release_root, &version_path, @@ -1589,6 +1763,15 @@ pub fn read_localized_version_state( pub fn write_localized_version_state( localized_output_root: &Path, state: &LocalizedVersionState, +) -> anyhow::Result { + let _lock = LocalizedOutputLock::acquire(localized_output_root)?; + recover_localized_transaction(localized_output_root)?; + write_localized_version_state_unlocked(localized_output_root, state) +} + +fn write_localized_version_state_unlocked( + localized_output_root: &Path, + state: &LocalizedVersionState, ) -> anyhow::Result { ensure_safe_directory_path(localized_output_root, "汉化输出目录") .map_err(anyhow::Error::msg)?; @@ -1609,6 +1792,8 @@ pub fn mark_localized_manual_proofreading( localized_output_root: &Path, official_release_id: &str, ) -> anyhow::Result { + let _lock = LocalizedOutputLock::acquire(localized_output_root)?; + recover_localized_transaction(localized_output_root)?; let mut state = read_localized_version_state(localized_output_root)?.unwrap_or_else(|| { LocalizedVersionState { state_version: LOCALIZED_VERSION_STATE_VERSION, @@ -1630,7 +1815,7 @@ pub fn mark_localized_manual_proofreading( state.translation_workflow_status = Some(LOCALIZED_TRANSLATION_STATUS_MANUAL_PROOFREADING.to_string()); state.updated_unix_seconds = unix_seconds_now(); - let state_path = write_localized_version_state(localized_output_root, &state)?; + let state_path = write_localized_version_state_unlocked(localized_output_root, &state)?; Ok(LocalizedTranslationWorkflowReport { command: "translation-proofread", @@ -1956,6 +2141,193 @@ fn verify_published_localized_release( Ok(integrity) } +fn build_localized_distribution_manifest( + official_release_root: &Path, + staging_root: &Path, + config: &LocalizedPatchConfig, +) -> anyhow::Result> { + let Some(official_manifest) = + crate::official_download::read_download_manifest_at(official_release_root) + .map_err(anyhow::Error::msg)? + else { + return Ok(None); + }; + let mut entries = Vec::with_capacity(official_manifest.entries.len()); + for entry in official_manifest.entries.values() { + let path = staging_root.join(&entry.destination); + ensure_path_within_root(staging_root, &path).map_err(anyhow::Error::msg)?; + ensure_safe_file_target(staging_root, &path, "localized distribution 文件") + .map_err(anyhow::Error::msg)?; + let bytes = fs::read(&path)?; + entries.push(LocalizedDistributionEntry { + url: entry.url.clone(), + destination: entry.destination.clone(), + bytes: bytes.len() as u64, + blake3: blake3::hash(&bytes).to_hex().to_string(), + }); + } + Ok(Some(LocalizedDistributionManifest { + version: 1, + official_release_id: config.release_id.clone(), + localized_release_id: config.published_release_id().to_string(), + entries, + })) +} + +fn verify_localized_release_files( + version_path: &Path, + manifest: &LocalizedPatchManifest, +) -> anyhow::Result<()> { + ensure_safe_directory_path(version_path, "localized release").map_err(anyhow::Error::msg)?; + for file in &manifest.files { + let path = version_path.join(&file.path); + ensure_path_within_root(version_path, &path).map_err(anyhow::Error::msg)?; + ensure_safe_file_target(version_path, &path, "localized release 文件") + .map_err(anyhow::Error::msg)?; + let bytes = fs::read(&path)?; + let actual = blake3::hash(&bytes).to_hex().to_string(); + if bytes.len() as u64 != file.localized_bytes || actual != file.localized_blake3 { + return Err(anyhow::anyhow!( + "localized release 文件完整性失败 {}：expected bytes={} blake3={} actual bytes={} blake3={}", + file.path, + file.localized_bytes, + file.localized_blake3, + bytes.len(), + actual + )); + } + } + Ok(()) +} + +fn write_localized_transaction( + localized_output_root: &Path, + transaction: &LocalizedReleaseTransaction, +) -> anyhow::Result<()> { + let path = localized_output_root.join(LOCALIZED_TRANSACTION_FILE); + write_file_atomic( + &path, + &serde_json::to_vec_pretty(transaction)?, + STATE_FILE_MODE, + "localized release transaction", + ) + .map_err(anyhow::Error::msg) +} + +fn update_localized_transaction_phase( + localized_output_root: &Path, + phase: &str, +) -> anyhow::Result<()> { + let path = localized_output_root.join(LOCALIZED_TRANSACTION_FILE); + let Some(bytes) = + read_file_no_symlink(&path, "localized release transaction").map_err(anyhow::Error::msg)? + else { + return Err(anyhow::anyhow!( + "localized release transaction 丢失：{}", + path.display() + )); + }; + let mut transaction: LocalizedReleaseTransaction = serde_json::from_slice(&bytes)?; + transaction.phase = phase.to_string(); + write_localized_transaction(localized_output_root, &transaction) +} + +fn remove_localized_transaction(localized_output_root: &Path) -> anyhow::Result<()> { + let path = localized_output_root.join(LOCALIZED_TRANSACTION_FILE); + match fs::symlink_metadata(&path) { + Ok(metadata) if metadata.file_type().is_symlink() => { + Err(anyhow::anyhow!("localized transaction 不能是 symlink")) + } + Ok(_) => { + fs::remove_file(path)?; + Ok(()) + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error.into()), + } +} + +fn recover_localized_transaction(localized_output_root: &Path) -> anyhow::Result<()> { + let path = localized_output_root.join(LOCALIZED_TRANSACTION_FILE); + let Some(bytes) = + read_file_no_symlink(&path, "localized release transaction").map_err(anyhow::Error::msg)? + else { + return Ok(()); + }; + let transaction: LocalizedReleaseTransaction = serde_json::from_slice(&bytes)?; + if transaction.version != 1 { + return Err(anyhow::anyhow!( + "不支持的 localized transaction schema：{}", + transaction.version + )); + } + ensure_path_within_root(localized_output_root, &transaction.version_path) + .map_err(anyhow::Error::msg)?; + ensure_safe_directory_path(&transaction.version_path, "localized transaction release") + .map_err(anyhow::Error::msg)?; + let current_path = localized_output_root.join(LOCALIZED_CURRENT_LINK); + let current_matches = match transaction.current_target.as_deref() { + Some(target) => current_path + .read_link() + .map(|current| current == target) + .unwrap_or(false), + None => matches!( + fs::symlink_metadata(¤t_path), + Err(error) if error.kind() == std::io::ErrorKind::NotFound + ), + }; + let state_matches = transaction.new_state.as_ref().is_some_and(|expected| { + read_localized_version_state(localized_output_root) + .ok() + .flatten() + .as_ref() + == Some(expected) + }); + let publish_committed = transaction.operation == "publish" + && transaction.version_path.is_dir() + && current_matches + && read_localized_version_state(localized_output_root) + .ok() + .flatten() + .and_then(|state| state.current_release_id) + .is_some_and(|id| id == transaction.release_id); + let rollback_committed = + transaction.operation == "rollback" && current_matches && state_matches; + + if publish_committed { + if let Some(staging) = transaction.staging_path.as_deref() { + remove_owned_path(staging)?; + } + } else if rollback_committed { + remove_owned_path(&transaction.version_path)?; + } else { + if let Some(target) = transaction.previous_current_target.as_deref() { + ensure_path_within_root(localized_output_root, target).map_err(anyhow::Error::msg)?; + restore_current_symlink(localized_output_root, ¤t_path, Some(target))?; + } else if transaction.operation == "publish" { + restore_current_symlink(localized_output_root, ¤t_path, None)?; + } + if transaction.operation == "publish" { + if let Some(staging) = transaction.staging_path.as_deref() { + remove_owned_path(staging)?; + } + remove_owned_path(&transaction.version_path)?; + } + if let Some(previous_state) = transaction.previous_state_bytes.as_deref() { + write_file_atomic( + &localized_output_root.join(LOCALIZED_VERSION_STATE_FILE), + previous_state, + STATE_FILE_MODE, + "恢复 localized version state", + ) + .map_err(anyhow::Error::msg)?; + } else { + remove_owned_path(&localized_output_root.join(LOCALIZED_VERSION_STATE_FILE))?; + } + } + remove_localized_transaction(localized_output_root) +} + /// Inspects one localized release without changing state, staging, current or /// any repair target. pub fn inspect_localized_release_artifact( @@ -2805,7 +3177,11 @@ fn rollback_failed_publish( .map(|current_target| current_target == *target) .unwrap_or(false) }) { - restore_current_symlink(localized_output_root, current_path, previous_current_target)?; + restore_current_symlink( + localized_output_root, + current_path, + previous_current_target.map(PathBuf::as_path), + )?; } remove_owned_path(state_path)?; if let Some(previous_state_bytes) = previous_state_bytes { @@ -2835,7 +3211,7 @@ fn remove_owned_path(path: &Path) -> anyhow::Result<()> { fn restore_current_symlink( root: &Path, current_path: &Path, - previous_current_target: Option<&PathBuf>, + previous_current_target: Option<&Path>, ) -> anyhow::Result<()> { use std::os::unix::fs::symlink; @@ -2857,7 +3233,7 @@ fn restore_current_symlink( fn restore_current_symlink( _root: &Path, _current_path: &Path, - _previous_current_target: Option<&PathBuf>, + _previous_current_target: Option<&Path>, ) -> anyhow::Result<()> { Ok(()) } @@ -3280,17 +3656,55 @@ mod tests { fs::create_dir_all(&target).unwrap(); let binary_source = b"binary-before"; - let binary_target = b"binary-after"; + let binary_target = b"binary-after-longer"; let json_source = br#"{"value":0}"#; let json_target = br#"{"value":1}"#; let text_source = "old text\n"; - let text_target = "new text\n"; + let text_target = "translated text with a different length\n"; fs::write(official.join("data.bin"), binary_source).unwrap(); fs::write(target.join("data.bin"), binary_target).unwrap(); fs::write(official.join("data.json"), json_source).unwrap(); fs::write(target.join("data.json"), json_target).unwrap(); fs::write(official.join("text.txt"), text_source).unwrap(); fs::write(target.join("text.txt"), text_target).unwrap(); + fs::write( + official.join("official-download-manifest.json"), + serde_json::to_vec(&crate::OfficialDownloadManifest { + version: 1, + entries: [ + ( + "https://example.invalid/data.bin".to_string(), + "data.bin", + binary_source.as_slice(), + ), + ( + "https://example.invalid/data.json".to_string(), + "data.json", + json_source.as_slice(), + ), + ( + "https://example.invalid/text.txt".to_string(), + "text.txt", + text_source.as_bytes(), + ), + ] + .into_iter() + .map(|(url, destination, bytes)| { + ( + url.clone(), + crate::OfficialDownloadManifestEntry { + url, + destination: destination.to_string(), + bytes: bytes.len() as u64, + blake3: blake3::hash(bytes).to_hex().to_string(), + }, + ) + }) + .collect(), + }) + .unwrap(), + ) + .unwrap(); let manifest = bat_patch::build_patch_manifest( &official, @@ -3389,6 +3803,29 @@ mod tests { ] ); assert!(report.integrity.current_points_to_release); + let distribution: LocalizedDistributionManifest = serde_json::from_slice( + &fs::read( + report + .version_path + .join(LOCALIZED_DISTRIBUTION_MANIFEST_FILE), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(distribution.entries.len(), 3); + for (path, expected) in [ + ("data.bin", binary_target.as_slice()), + ("data.json", json_target.as_slice()), + ("text.txt", text_target.as_bytes()), + ] { + let entry = distribution + .entries + .iter() + .find(|entry| entry.destination == path) + .unwrap(); + assert_eq!(entry.bytes, expected.len() as u64); + assert_eq!(entry.blake3, blake3::hash(expected).to_hex().to_string()); + } } #[cfg(unix)] diff --git a/infrastructure/src/official_download.rs b/infrastructure/src/official_download.rs index 8d5b885..dc9997d 100644 --- a/infrastructure/src/official_download.rs +++ b/infrastructure/src/official_download.rs @@ -393,12 +393,25 @@ pub fn read_cas_reuse_reference_manifest_at( /// Decrements and removes CAS references recorded for a release. /// -/// The operation is resumable: after every successful decrement the remaining -/// object IDs are atomically written back to the release-local manifest. +/// Each decrement is committed together with a durable `(release, ordinal)` +/// ownership record in CAS metadata. The release-local manifest remains a +/// resumable progress cursor, so a crash before its rewrite cannot decrement +/// the same ownership twice. pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Result { let Some(mut manifest) = read_cas_reuse_reference_manifest_at(release_root)? else { return Ok(0); }; + let release_id = release_root + .file_name() + .and_then(|name| name.to_str()) + .filter(|name| !name.is_empty() && *name != "." && *name != "..") + .ok_or_else(|| { + format!( + "无法从 release 路径确定 CAS ownership ID：{}", + release_root.display() + ) + })? + .to_string(); let objects_root = cas_root.join("objects"); let metadata_path = cas_root.join("metadata.sqlite"); require_existing_directory(cas_root, "CAS 根目录")?; @@ -406,20 +419,24 @@ pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Res require_existing_file(cas_root, &metadata_path, "CAS 元数据库")?; let mut released = 0usize; while let Some(object_id) = manifest.object_ids.pop() { + let ordinal = manifest.object_ids.len() as u64; let cas_root = cas_root.to_path_buf(); let object_id_for_runtime = object_id.clone(); + let release_id_for_runtime = release_id.clone(); let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build() .map_err(|error| format!("创建 CAS 引用清理 runtime 失败：{error}"))?; - runtime.block_on(async move { + let did_release = runtime.block_on(async move { let cas = crate::FileSystemCasRepository::new(cas_root); - cas.remove_reference(&object_id_for_runtime) + cas.release_reference_once(&release_id_for_runtime, ordinal, &object_id_for_runtime) .await .map_err(|error| format!("减少 CAS release 引用失败 object={object_id}：{error}")) })?; write_cas_reuse_reference_manifest(release_root, &manifest)?; - released += 1; + if did_release { + released += 1; + } } let path = release_root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE); match fs::symlink_metadata(&path) { @@ -4656,6 +4673,26 @@ exit 22 assert!(read_cas_reuse_reference_manifest_at(&out_dir) .unwrap() .is_none()); + + // 模拟 CAS 事务已提交但 release-local progress cursor 尚未写回； + // 重试必须识别同一个 ownership pair，而不是再次递减。 + fs::write( + out_dir.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE), + serde_json::to_vec(&OfficialCasReuseReferenceManifest { + version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION, + object_ids: vec![object_id.clone()], + }) + .unwrap(), + ) + .unwrap(); + assert_eq!( + release_cas_reuse_references(&out_dir, &cas_root).unwrap(), + 0 + ); + assert_eq!(cas_reference_count(&cas_root, &object_id), 1); + assert!(read_cas_reuse_reference_manifest_at(&out_dir) + .unwrap() + .is_none()); } #[test] diff --git a/infrastructure/src/official_update.rs b/infrastructure/src/official_update.rs index 1e2cbe2..f7fe840 100644 --- a/infrastructure/src/official_update.rs +++ b/infrastructure/src/official_update.rs @@ -3224,7 +3224,15 @@ fn copy_tree_no_symlink( continue; } } - if let Err(_error) = fs::hard_link(&source_path, &destination_path) { + if is_release_local_mutable_state(&source_path) { + fs::copy(&source_path, &destination_path).map_err(|copy_error| { + format!( + "复制官方 release mutable state 失败 {} -> {}：{copy_error}", + source_path.display(), + destination_path.display() + ) + })?; + } else if let Err(_error) = fs::hard_link(&source_path, &destination_path) { fs::copy(&source_path, &destination_path).map_err(|copy_error| { format!( "复制官方资源到 staging 失败 {} -> {}：{copy_error}", @@ -3243,6 +3251,17 @@ fn copy_tree_no_symlink( Ok(()) } +fn is_release_local_mutable_state(path: &Path) -> bool { + matches!( + path.file_name().and_then(|name| name.to_str()), + Some( + "translation-tasks.sqlite" + | "translation-tasks.sqlite-wal" + | "translation-tasks.sqlite-shm" + ) + ) +} + #[cfg(unix)] fn switch_current_symlink( root: &Path, @@ -4415,6 +4434,46 @@ fn process_exists(_pid: u32) -> bool { mod tests { use super::*; + #[cfg(unix)] + #[test] + fn staging_copy_does_not_hard_link_mutable_translation_state() { + use std::os::unix::fs::MetadataExt; + + let temp = tempfile::TempDir::new().unwrap(); + let source = temp.path().join("old"); + let destination = temp.path().join("new"); + fs::create_dir_all(&source).unwrap(); + fs::write(source.join("translation-tasks.sqlite"), b"old-db").unwrap(); + fs::write(source.join("translation-tasks.sqlite-wal"), b"old-wal").unwrap(); + fs::write(source.join("immutable.bundle"), b"payload").unwrap(); + + copy_tree_no_symlink(&source, &destination, false).unwrap(); + + let old_db_inode = fs::metadata(source.join("translation-tasks.sqlite")) + .unwrap() + .ino(); + let new_db_inode = fs::metadata(destination.join("translation-tasks.sqlite")) + .unwrap() + .ino(); + assert_ne!(old_db_inode, new_db_inode); + fs::write(destination.join("translation-tasks.sqlite"), b"new-db").unwrap(); + fs::write(destination.join("translation-tasks.sqlite-wal"), b"new-wal").unwrap(); + assert_eq!( + fs::read(source.join("translation-tasks.sqlite")).unwrap(), + b"old-db" + ); + assert_eq!( + fs::read(source.join("translation-tasks.sqlite-wal")).unwrap(), + b"old-wal" + ); + assert_eq!( + fs::metadata(source.join("immutable.bundle")).unwrap().ino(), + fs::metadata(destination.join("immutable.bundle")) + .unwrap() + .ino() + ); + } + #[test] fn missing_app_version_carries_input_error_code() { // 未启用 auto-discover 且未传 app-version：配置校验失败应携带 diff --git a/infrastructure/src/release_ops.rs b/infrastructure/src/release_ops.rs index 10d2cbc..de5cb1a 100644 --- a/infrastructure/src/release_ops.rs +++ b/infrastructure/src/release_ops.rs @@ -6,8 +6,8 @@ use crate::localized_patch::{ inspect_localized_release_artifact_at, read_localized_patch_manifest_at, - read_localized_version_state, LOCALIZED_CURRENT_LINK, LOCALIZED_STAGING_DIR, - LOCALIZED_VERSIONS_DIR, + read_localized_version_state, LocalizedDistributionManifest, LOCALIZED_CURRENT_LINK, + LOCALIZED_DISTRIBUTION_MANIFEST_FILE, LOCALIZED_STAGING_DIR, LOCALIZED_VERSIONS_DIR, }; use crate::official_download::{ read_cas_reuse_reference_manifest_at, read_download_manifest_at, release_cas_reuse_references, @@ -48,6 +48,9 @@ pub struct ReleaseDistributionParams { /// Entry page size. Zero uses the server default. #[serde(default)] pub limit: usize, + /// Optional single destination whose bytes are revalidated. + #[serde(default)] + pub destination: Option, } /// Parameters for the two-step cleanup operation. @@ -342,49 +345,25 @@ pub fn select_release_distribution( official_root: &Path, localized_root: &Path, params: &ReleaseDistributionParams, - unzip_command: &Path, + _unzip_command: &Path, ) -> anyhow::Result { let channel = Channel::parse(params.channel.as_deref())?; - let status = build_release_status(official_root, localized_root, unzip_command)?; let selected_id = params.release_id.as_deref(); - let selected = status.releases.iter().find(|release| { - release.channel == channel.as_str() - && selected_id.is_none_or(|id| release.id == id) - && (selected_id.is_some() || release.current) - }); - let Some(selected) = selected else { + let Some(selection) = select_release_distribution_metadata( + official_root, + localized_root, + channel, + selected_id, + params.destination.as_deref(), + )? + else { return Ok(blocked_distribution( channel, selected_id.map(str::to_string), - "请求的 release 不存在或不是当前 release", + "请求的 release 不存在、publication identity 无效或不是当前 release", )); }; - if selected.distribution_integrity_status != "valid" { - return Ok(blocked_distribution( - channel, - Some(selected.id.clone()), - "release 产物完整性未通过，拒绝分发", - )); - } - let root = selected.path.clone(); - let manifest_root = selected - .source_official_release_id - .as_deref() - .map(|source| official_root.join(OFFICIAL_VERSIONS_DIR).join(source)) - .unwrap_or_else(|| root.clone()); - let manifest = read_download_manifest_at(&manifest_root) - .map_err(anyhow::Error::msg)? - .ok_or_else(|| anyhow::anyhow!("release 缺少官方下载 manifest"))?; - let all_entries = manifest - .entries - .values() - .map(|entry| ReleaseDistributionEntry { - url: entry.url.clone(), - destination: entry.destination.clone(), - bytes: entry.bytes, - blake3: entry.blake3.clone(), - }) - .collect::>(); + let all_entries = selection.entries; let total = all_entries.len(); let offset = params.offset.min(total); let limit = if params.limit == 0 { @@ -396,22 +375,212 @@ pub fn select_release_distribution( Ok(ReleaseDistributionPage { available: true, channel: channel.as_str().to_string(), - release_id: Some(selected.id.clone()), - resource_root: Some(root), - source_official_release_id: selected.source_official_release_id.clone(), - current: selected.current, + release_id: Some(selection.id), + resource_root: Some(selection.path), + source_official_release_id: selection.source_official_release_id, + current: selection.current, status: ReleaseFlowStatusCode::DistributionReady .status() .to_string(), status_code: ReleaseFlowStatusCode::DistributionReady .as_str() .to_string(), - artifact_integrity_status: selected.artifact_integrity_status.clone(), + artifact_integrity_status: "valid".to_string(), total, offset, limit, entries, - diagnostics: selected.diagnostics.clone(), + diagnostics: Vec::new(), + }) +} + +struct DistributionMetadataSelection { + id: String, + path: PathBuf, + source_official_release_id: Option, + current: bool, + entries: Vec, +} + +fn select_release_distribution_metadata( + official_root: &Path, + localized_root: &Path, + channel: Channel, + selected_id: Option<&str>, + destination: Option<&str>, +) -> anyhow::Result> { + let (root, versions_dir, current_link) = match channel { + Channel::Official => (official_root, OFFICIAL_VERSIONS_DIR, OFFICIAL_CURRENT_LINK), + Channel::Localized => ( + localized_root, + LOCALIZED_VERSIONS_DIR, + LOCALIZED_CURRENT_LINK, + ), + }; + ensure_safe_directory_path(root, "release distribution 根目录").map_err(anyhow::Error::msg)?; + let current_id = match channel { + Channel::Official => read_version_state(&official_root.join(OFFICIAL_VERSION_STATE_FILE))? + .and_then(|state| state.current_completed_version.map(|record| record.id)), + Channel::Localized => { + read_localized_version_state(localized_root)?.and_then(|state| state.current_release_id) + } + }; + let requested_id = selected_id.or(current_id.as_deref()); + let Some(id) = requested_id else { + return Ok(None); + }; + if !is_safe_release_id(id) { + return Ok(None); + } + let path = root.join(versions_dir).join(id); + ensure_path_within_root(root, &path).map_err(anyhow::Error::msg)?; + ensure_safe_directory_path(&path, "release distribution version") + .map_err(anyhow::Error::msg)?; + if !path.is_dir() { + return Ok(None); + } + let pointer_id = read_managed_current_id(&root.join(current_link), versions_dir); + let current = current_id.as_deref() == Some(id) && pointer_id.as_deref() == Some(id); + if selected_id.is_none() && !current { + return Ok(None); + } + + match channel { + Channel::Official => { + let manifest = read_download_manifest_at(&path) + .map_err(anyhow::Error::msg)? + .ok_or_else(|| anyhow::anyhow!("official release 缺少官方下载 manifest"))?; + let entries = manifest + .entries + .values() + .map(|entry| ReleaseDistributionEntry { + url: entry.url.clone(), + destination: entry.destination.clone(), + bytes: entry.bytes, + blake3: entry.blake3.clone(), + }) + .collect::>(); + if !validate_distribution_entries(&path, &entries, destination)? { + return Ok(None); + } + Ok(Some(DistributionMetadataSelection { + id: id.to_string(), + path, + source_official_release_id: None, + current, + entries, + })) + } + Channel::Localized => { + let bytes = crate::path_security::read_file_no_symlink( + &path.join(LOCALIZED_DISTRIBUTION_MANIFEST_FILE), + "localized distribution manifest", + ) + .map_err(anyhow::Error::msg)? + .ok_or_else(|| anyhow::anyhow!("localized release 缺少 distribution manifest"))?; + let manifest: LocalizedDistributionManifest = serde_json::from_slice(&bytes)?; + if manifest.version != 1 + || manifest.localized_release_id != id + || !is_safe_release_id(&manifest.official_release_id) + { + return Ok(None); + } + let source_path = official_root + .join(OFFICIAL_VERSIONS_DIR) + .join(&manifest.official_release_id); + ensure_safe_directory_path(&source_path, "localized source official release") + .map_err(anyhow::Error::msg)?; + let Some(official_manifest) = (if source_path.is_dir() { + read_download_manifest_at(&source_path).map_err(anyhow::Error::msg)? + } else { + None + }) else { + return Ok(None); + }; + if !localized_distribution_matches_official(&manifest, &official_manifest) { + return Ok(None); + } + let entries = manifest + .entries + .into_iter() + .map(|entry| ReleaseDistributionEntry { + url: entry.url, + destination: entry.destination, + bytes: entry.bytes, + blake3: entry.blake3, + }) + .collect::>(); + if !validate_distribution_entries(&path, &entries, destination)? { + return Ok(None); + } + Ok(Some(DistributionMetadataSelection { + id: id.to_string(), + path, + source_official_release_id: Some(manifest.official_release_id), + current, + entries, + })) + } + } +} + +fn validate_distribution_entries( + root: &Path, + entries: &[ReleaseDistributionEntry], + destination: Option<&str>, +) -> anyhow::Result { + for entry in entries { + let path = root.join(&entry.destination); + ensure_path_within_root(root, &path).map_err(anyhow::Error::msg)?; + if destination.is_none() { + let metadata = match fs::symlink_metadata(&path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(error.into()), + }; + if !metadata.is_file() + || metadata.file_type().is_symlink() + || metadata.len() != entry.bytes + { + return Ok(false); + } + } + } + let Some(destination) = destination else { + return Ok(true); + }; + let Some(entry) = entries + .iter() + .find(|entry| entry.destination == destination) + else { + return Ok(false); + }; + let path = root.join(destination); + ensure_safe_file_target(root, &path, "release distribution 文件") + .map_err(anyhow::Error::msg)?; + let bytes = fs::read(&path)?; + Ok(bytes.len() as u64 == entry.bytes + && blake3::hash(&bytes).to_hex().to_string() == entry.blake3) +} + +fn localized_distribution_matches_official( + localized: &LocalizedDistributionManifest, + official: &OfficialDownloadManifest, +) -> bool { + if localized.entries.len() != official.entries.len() { + return false; + } + let mut localized_by_destination = BTreeMap::new(); + for entry in &localized.entries { + if localized_by_destination + .insert(entry.destination.as_str(), entry.url.as_str()) + .is_some() + { + return false; + } + } + official.entries.values().all(|entry| { + localized_by_destination.get(entry.destination.as_str()) == Some(&entry.url.as_str()) }) } @@ -615,6 +784,13 @@ fn build_cleanup_plan( &mut entries, &mut diagnostics, )?; + entries.sort_by(|left, right| { + left.channel + .cmp(&right.channel) + .then_with(|| left.id.cmp(&right.id)) + .then_with(|| left.path.cmp(&right.path)) + }); + diagnostics.sort(); let fingerprint = serde_json::to_vec(&(&entries, &diagnostics))?; let plan_id = blake3::hash(&fingerprint).to_hex().to_string(); Ok(CleanupPlan { diff --git a/internal/api/cdn.go b/internal/api/cdn.go index 01d2934..bb24a63 100644 --- a/internal/api/cdn.go +++ b/internal/api/cdn.go @@ -33,7 +33,7 @@ func (s *Server) serveCDN(w http.ResponseWriter, r *http.Request) { http.Error(w, selectorErr.Error(), http.StatusBadRequest) return } - selected, selectErr := s.loadReleaseDistribution(r, channel, releaseID) + selected, selectErr := s.loadReleaseDistribution(r, channel, releaseID, rel) if selectErr != nil { http.Error(w, selectErr.Error(), http.StatusServiceUnavailable) return @@ -72,7 +72,11 @@ func (s *Server) serveCDN(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) return } - if (explicitRelease || s.cfg.RequireIndexed) && s.cfg.VerifySize { + if explicitRelease && hasEntry && uint64(info.Size()) != entry.Bytes { + http.Error(w, "size mismatch with release index", http.StatusConflict) + return + } + if !explicitRelease && s.cfg.RequireIndexed && s.cfg.VerifySize { if hasEntry && entry.Bytes > 0 && uint64(info.Size()) != entry.Bytes { http.Error(w, "size mismatch with release index", http.StatusConflict) return diff --git a/internal/api/openapi.go b/internal/api/openapi.go index a7616dd..8f15182 100644 --- a/internal/api/openapi.go +++ b/internal/api/openapi.go @@ -107,6 +107,11 @@ paths: in: query schema: type: string + - name: destination + in: query + description: Optional release-relative path whose localized bytes and BLAKE3 are revalidated. + schema: + type: string - name: offset in: query schema: diff --git a/internal/api/release_http.go b/internal/api/release_http.go index 2d92379..73af42f 100644 --- a/internal/api/release_http.go +++ b/internal/api/release_http.go @@ -139,12 +139,13 @@ func (e *releaseSelectorError) Error() string { return e.message } -func (s *Server) loadReleaseDistribution(r *http.Request, channel, releaseID string) (*backendrpc.ReleaseDistributionPage, error) { +func (s *Server) loadReleaseDistribution(r *http.Request, channel, releaseID, destination string) (*backendrpc.ReleaseDistributionPage, error) { return s.loadReleaseDistributionFrom(r, backendrpc.ReleaseDistributionParams{ - Channel: channel, - ReleaseID: releaseID, - Offset: 0, - Limit: 1000, + Channel: channel, + ReleaseID: releaseID, + Destination: destination, + Offset: 0, + Limit: 1000, }) } @@ -154,6 +155,9 @@ func releaseDistributionParams(r *http.Request, channel, releaseID string) (back ReleaseID: releaseID, } query := r.URL.Query() + if destination := strings.TrimSpace(query.Get("destination")); destination != "" { + params.Destination = destination + } if raw := strings.TrimSpace(query.Get("offset")); raw != "" { offset, err := strconv.ParseUint(raw, 10, 64) if err != nil || uint64(int(^uint(0)>>1)) < offset { @@ -196,10 +200,11 @@ func (s *Server) loadReleaseDistributionFrom(r *http.Request, params backendrpc. all := append([]backendrpc.ReleaseDistributionEntry(nil), result.Entries...) for offset := len(all); offset < result.Total; { next, nextErr := backend.ReleaseDistribution(r.Context(), backendrpc.ReleaseDistributionParams{ - Channel: params.Channel, - ReleaseID: params.ReleaseID, - Offset: offset, - Limit: pageSize, + Channel: params.Channel, + ReleaseID: params.ReleaseID, + Destination: params.Destination, + Offset: offset, + Limit: pageSize, }) if nextErr != nil { return nil, nextErr diff --git a/internal/api/release_http_test.go b/internal/api/release_http_test.go index 22062d4..eeba4d4 100644 --- a/internal/api/release_http_test.go +++ b/internal/api/release_http_test.go @@ -101,13 +101,14 @@ func TestReleaseHTTPForwardsTypedSelectionAndDoesNotFallback(t *testing.T) { } recorder = httptest.NewRecorder() - server.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/v1/distribution?channel=localized&release_id=localized-1&offset=2&limit=10", nil)) + server.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/v1/distribution?channel=localized&release_id=localized-1&destination=TableBundles%2FTableCatalog.bytes&offset=2&limit=10", nil)) if recorder.Code != http.StatusOK { t.Fatalf("distribution status=%d body=%s", recorder.Code, recorder.Body.String()) } if len(backend.distributionParams) != 1 || backend.distributionParams[0].Channel != "localized" || backend.distributionParams[0].ReleaseID != "localized-1" || + backend.distributionParams[0].Destination != "TableBundles/TableCatalog.bytes" || backend.distributionParams[0].Offset != 2 || backend.distributionParams[0].Limit != 10 { t.Fatalf("distribution params=%#v", backend.distributionParams) diff --git a/internal/backendrpc/client.go b/internal/backendrpc/client.go index 8520609..4388ec9 100644 --- a/internal/backendrpc/client.go +++ b/internal/backendrpc/client.go @@ -189,10 +189,11 @@ type ReleaseListParams struct { // ReleaseDistributionParams selects a verified release for distribution. type ReleaseDistributionParams struct { - Channel string `json:"channel,omitempty"` - ReleaseID string `json:"release_id,omitempty"` - Offset int `json:"offset,omitempty"` - Limit int `json:"limit,omitempty"` + Channel string `json:"channel,omitempty"` + ReleaseID string `json:"release_id,omitempty"` + Offset int `json:"offset,omitempty"` + Limit int `json:"limit,omitempty"` + Destination string `json:"destination,omitempty"` } // ReleaseCleanupParams controls the dry-run/execute cleanup pair.