diff --git a/CURRENT_STATUS.md b/CURRENT_STATUS.md index 22f0826..8dca161 100644 --- a/CURRENT_STATUS.md +++ b/CURRENT_STATUS.md @@ -51,11 +51,14 @@ TextUnit scope、source history 和 approved review。worker、TM 复用、人 `localized.degraded`，只读检查不会自动回滚、删除或修复。双 release 的 `release.status/list/distribution/cleanup` 已由 Rust 从既有状态、manifest、文件系统和 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。 +和 GC 通过跨进程操作锁协调，release-local CAS 引用以 `(ownership_id, ordinal)` ownership +记录幂等释放；新清单持久化 `ownership_id`，旧清单走不生成新 identity 的 legacy +cleanup path；localized publish/rollback 通过 output-root 单写者锁和事务日志恢复 +current、version-state、version 目录，publish 只有最终 `verified` phase 才能恢复为已提交。 +localized release 还写入实际 bytes/BLAKE3 的 `localized-distribution-manifest.json`， +`release.distribution(destination=...)` 使用单条轻量 metadata lookup，返回 exactly one +entry 且不在分发热路径执行完整 release audit；`release.cleanup execute` 与 official +sync 共用同一个 `.official-sync.lock`，localized cleanup 继续使用 `.localized-release.lock`。 --- diff --git a/api/openapi/bat-api.yaml b/api/openapi/bat-api.yaml index aad3e54..8b61834 100644 --- a/api/openapi/bat-api.yaml +++ b/api/openapi/bat-api.yaml @@ -102,7 +102,7 @@ paths: type: string - name: destination in: query - description: Optional release-relative path whose localized bytes and BLAKE3 are revalidated. + description: Optional release-relative path for single-entry lookup; Rust returns exactly one entry and revalidates the selected channel's actual bytes and BLAKE3. schema: type: string - name: offset diff --git a/crates/bat-cas-engine/src/refcount.rs b/crates/bat-cas-engine/src/refcount.rs index 918d634..9618fd4 100644 --- a/crates/bat-cas-engine/src/refcount.rs +++ b/crates/bat-cas-engine/src/refcount.rs @@ -303,12 +303,13 @@ impl SqliteRefCounter { /// 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. + /// same SQLite transaction. Retrying the same `(ownership_id, ordinal)` is + /// therefore idempotent, while a different ownership keeps its own row and + /// reference count. The legacy SQL column name is retained for schema + /// compatibility. pub async fn release_reference_once( &self, - release_id: &str, + ownership_id: &str, ordinal: u64, hash: &Hash, ) -> Result { @@ -320,7 +321,7 @@ impl SqliteRefCounter { WHERE release_id = ?1 AND ordinal = ?2 "#, ) - .bind(release_id) + .bind(ownership_id) .bind(ordinal as i64) .fetch_optional(&mut *transaction) .await?; @@ -328,8 +329,8 @@ impl SqliteRefCounter { 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, + "CAS release ownership mismatch: ownership_id={} ordinal={} expected={} actual={}", + ownership_id, ordinal, object_id, hash @@ -340,8 +341,8 @@ impl SqliteRefCounter { return Ok(false); } return Err(CasError::Other(anyhow::anyhow!( - "CAS release ownership record is not in a retryable state: release={} ordinal={}", - release_id, + "CAS release ownership record is not in a retryable state: ownership_id={} ordinal={}", + ownership_id, ordinal ))); } @@ -380,7 +381,7 @@ impl SqliteRefCounter { VALUES(?1, ?2, ?3, 1) "#, ) - .bind(release_id) + .bind(ownership_id) .bind(ordinal as i64) .bind(hash.to_string()) .execute(&mut *transaction) diff --git a/crates/bat-cas-engine/src/repository.rs b/crates/bat-cas-engine/src/repository.rs index 5784bec..ddc2b87 100644 --- a/crates/bat-cas-engine/src/repository.rs +++ b/crates/bat-cas-engine/src/repository.rs @@ -166,13 +166,13 @@ impl FileSystemCasRepository { /// Releases one release-owned reference exactly once. pub async fn release_reference_once( &self, - release_id: &str, + ownership_id: &str, ordinal: u64, hash: &Hash, ) -> Result { let _lock = self.acquire_operation_lock().await?; self.ref_counter - .release_reference_once(release_id, ordinal, hash) + .release_reference_once(ownership_id, ordinal, hash) .await } diff --git a/docs/architecture/official-resource-backend.md b/docs/architecture/official-resource-backend.md index c898c1d..8f820f8 100644 --- a/docs/architecture/official-resource-backend.md +++ b/docs/architecture/official-resource-backend.md @@ -243,7 +243,7 @@ trusted 和 release/TextUnit/provider/run provenance；`translation.tasks` 优 12. 将 staging rename 为 `/versions/`，再原子替换 `/current` symlink 指向该 versioned 目录。 13. 发布完成后先对比上一完整 release 和当前 release 的 `official-download-manifest.json`，写出 `official-resource-changes.json` 和 `crowdin-translation-handoff.json`。同一 destination 只有 size 或 BLAKE3 变化才算 modified；新增+变更资源进入解析/翻译 handoff，删除资源只进入差异记录。当前只预留 Crowdin 本地 handoff，不发外部 API 请求。 14. 随后刷新 active release 下的 `official-parse-cache.json` 和 `official-textunit-index.json`，并从 Added/Modified 资源、parse cache 与 TextUnit 明细索引派生 `official-textunit-tasks.json`、`crowdin-textunit-queue.json` 和版本化的 `translation-tasks.sqlite`；up-to-date 轮询在已有有效解析缓存、TextUnit 明细索引和 TextUnit 队列时只读取摘要，不重复解析，重新同步队列时保留已有 worker 状态。历史 release 复用只允许不可变资源 payload/sidecar 硬链接；download manifest、snapshot、parse/textunit cache、queue、handoff、bootstrap、CAS reuse references 以及 `translation-tasks.sqlite`、WAL/SHM 都必须独立复制，不能共享可变 inode。 -15. 若启用 `--import-repository`，已校验 release 会被导入 CAS + `ResourceRepository`，并可经 `resource.index` 查询。历史 release 候选失效时，已有 CAS 对象会先经过完整性和元数据校验，再增加 release 引用并原子物化；当前 release 在 `official-cas-reuse-references.json` 中记录引用，staging/release 清理时递减，失败则回退网络并保留诊断。 +15. 若启用 `--import-repository`，已校验 release 会被导入 CAS + `ResourceRepository`，并可经 `resource.index` 查询。历史 release 候选失效时，已有 CAS 对象会先经过完整性和元数据校验，再增加 release 引用并原子物化；当前 release 在 `official-cas-reuse-references.json` 中记录引用和首次生成后持久化的 `ownership_id`，staging/release 清理时按 `ownership_id + ordinal` 递减，旧无 identity 清单走明确 legacy key 路径，失败则回退网络并保留诊断。 16. 官方同步报告默认给出 `localized_release_status=not_localized`，表示原版资源已发布、汉化资源未发布；generic manifest 驱动的 Binary/JSON/Text 以及当前支持的 UnityFS TextAsset、TypeTree string field 和 managed-reference string field patch 发布成功并通过 `localized-patch-manifest.json`、current symlink、release ID 及 ZIP 内层最终重解析校验后，`localized.status` 才返回 `localized`，表示原版和汉化两套资源都已发布。`localized.status` 分开返回 `patch_manifest_contract_status` 与 `artifact_integrity_status`；state/current/identity 存在但文件被截断或手工修改时返回 `localized.degraded`，只读检查不回滚、不删除、不修复。`translation.proofread` 只会把 workflow 标记成 `manual_proofreading` / `translation.manual_proofreading`，不会回退已发布汉化 release 的发布状态。 维护期特殊分支：如果官方 launcher/server-info 已经指向新资源根，但 client-patch seed marker 或必需 seed catalog 仍返回 403/404 等未开放状态，`bat` 返回 `waiting_for_official_resources`，保留现有 `current`，不创建失败 staging；若本轮启用 `--auto-discover`，会在 `/official-launcher-bootstrap.pending.json` 写入待处理 launcher bootstrap 证据，供后续排障和自研客户端开发使用。 diff --git a/docs/architecture/resource-release-layout.md b/docs/architecture/resource-release-layout.md index 4e00233..c8c9911 100644 --- a/docs/architecture/resource-release-layout.md +++ b/docs/architecture/resource-release-layout.md @@ -38,7 +38,7 @@ crowdin-textunit-queue.json # Crowdin worker 离线输入队列 official-sync-snapshot.json # 常在 active root / current 下 official-launcher-bootstrap.json # 官方 launcher 引导链版本化产物 - official-cas-reuse-references.json # 当前 release 获取的 CAS 引用 + official-cas-reuse-references.json # 当前 release 获取的 CAS 引用和 ownership_id prod-clientpatch.bluearchiveyostar.com/ / TableBundles/ @@ -110,7 +110,10 @@ manifest 做轻量选择，HTTP 热路径不重新执行完整 release audit；l localized publish/rollback 先取得 `.localized-release.lock`，并在 output root 下记录 `.localized-transaction.json`。current、version-state 和 version 目录的切换按日志阶段 -推进；下一次写操作会先恢复或完成未决事务，避免跨进程并发写入和中断后留下半发布状态。 +推进；publish 只有最终 `verified` phase 才能 roll-forward，下一次写操作会先恢复或完成 +未决事务，避免跨进程并发写入和中断后留下半发布状态。`release.cleanup execute` 先取得 +同一 official `.official-sync.lock`，再按固定顺序取得 localized 锁并在持锁状态下重建 +计划；dry-run 不占用 official mutation lock。 `release.cleanup` 先生成 dry-run 计划和 `plan_id`，执行时重新计算并比对计划。current、 rollback previous、active/in-progress、localized source official、state/manifest/CAS @@ -132,9 +135,12 @@ reference、无法确认 ownership 的对象均保留；只删除重新验证后 跨文件系统时复制到 staging 内的临时文件并原子 rename，旧 `versions/` 目录 保持不可变。 -从 CAS 物化资源时，`official-cas-reuse-references.json` 记录每个获取的对象引用， -文件带版本字段且允许重复 object ID。孤儿 staging 或显式 release 清理必须先按 -清单减少 CAS 引用，再删除目录；CAS 对象损坏、缺失或元数据不一致时只产生诊断， +从 CAS 物化资源时，`official-cas-reuse-references.json` 首次创建时生成并持久化 +`ownership_id`，记录每个获取的对象引用；文件带版本字段且允许重复 object ID。 +没有 `ownership_id` 的旧清单保留 legacy ownership key，不在读取时随机迁移。 +孤儿 staging 或显式 release 清理必须先按 +清单减少 CAS 引用，再删除目录；cleanup execute 与官方同步共用 +`.official-sync.lock`，localized cleanup 使用 `.localized-release.lock`。CAS 对象损坏、缺失或元数据不一致时只产生诊断， 回退网络下载，不发布未经校验的文件。 --- diff --git a/docs/guides/official-resource-test-pull.md b/docs/guides/official-resource-test-pull.md index 80db9dc..d090d65 100644 --- a/docs/guides/official-resource-test-pull.md +++ b/docs/guides/official-resource-test-pull.md @@ -185,7 +185,7 @@ cargo run -p bat-infrastructure --example official_pull_plan -- \ - 新下载先写 `.part`，成功并通过必要校验后再替换为最终文件；如果断点续传后的 `.zip` 结构校验失败，会删除 `.part` 并重新全量下载 - 如果上一轮非 dry-run 已进入 staging 但未发布成功，下一轮会优先查找 `/official-version-state.json` 中同一 app version、bundle version 和 Addressables root 的失败版本；只有对应 `/.staging/` 仍存在、路径安全且 `versions/` 尚未发布时，才复用该 staging，并继续按 manifest 校验复用或重下单个 URL - 新 release 的 staging 在访问网络前会扫描已发布 release 的 `official-download-manifest.json`。候选必须同时满足 manifest 记录的 destination、size、BLAKE3 和适用的 ZIP 结构校验；URL、CDN 根和 release ID 的变化本身不会阻止复用。命中后优先用硬链接，跨文件系统时回退为临时文件复制并原子 rename，旧 release 不会被修改 -- 历史 release 候选失效时，如果配置的 CAS 根已有对应 BLAKE3 对象，会先通过 CAS 读取完整性和元数据，再增加当前 release 的引用并原子物化；当前 release 会写 `official-cas-reuse-references.json`，清理孤儿 staging 或显式清理 release 时递减这些引用。CAS 损坏、缺对象或元数据不一致会写入复用诊断并继续走网络下载，不会静默使用缓存 +- 历史 release 候选失效时，如果配置的 CAS 根已有对应 BLAKE3 对象，会先通过 CAS 读取完整性和元数据，再增加当前 release 的引用并原子物化；当前 release 会写带持久化 `ownership_id` 的 `official-cas-reuse-references.json`，清理孤儿 staging 或显式清理 release 时按 ownership 和 ordinal 递减这些引用；旧无 identity 清单保留 legacy cleanup key。CAS 损坏、缺对象或元数据不一致会写入复用诊断并继续走网络下载，不会静默使用缓存 - 把结果发布到 `--output/current` ## 5. 自动更新检查 @@ -229,7 +229,7 @@ cargo run -p bat-infrastructure --example official_pull_plan -- \ - `/official-bootstrap-cache.json`：`--auto-discover` 的 `GameMainConfig` 解析缓存。launcher metadata 与 remote manifest 文件列表 digest 都未变时复用缓存；任一变化时才通过官方 HTTP 按 manifest 下载必要 `resources.assets` 或旧版 game zip 到临时目录解析。 - `/official-version-state.json`：资源发布根目录的持久版本状态，包含当前已完成版本、正在拉取版本、上一个可用版本和失败版本。 - `/current/official-download-manifest.json`：本地下载强校验清单，记录 URL、相对路径、size 和 BLAKE3。 -- `/current/official-cas-reuse-references.json`：当前 release 获取的 CAS 引用清单；每个复用项占一条记录，release 清理或孤儿 staging GC 时据此递减引用。 +- `/current/official-cas-reuse-references.json`：当前 release 获取的 CAS 引用清单，首次创建时包含持久化 `ownership_id`；每个复用项占一条记录，release 清理或孤儿 staging GC 时据此按 ownership/ordinal 递减引用。 - `/current/official-resource-changes.json`：当前 release 相对上一完整 release 的资源差异，记录新增、变更、删除以及解析/翻译候选计数。 - `/current/crowdin-translation-handoff.json`：为后续 Crowdin worker 预留的本地队列，只包含新增+变更资源；它不是 Crowdin API 调用结果。 - `/current/official-parse-cache.json`：官方资源发布后的派生解析缓存，记录 bundle/zip 条目解析摘要和缓存复用情况；它不是汉化产物。 @@ -326,7 +326,7 @@ cargo run -p bat-infrastructure --bin bat -- \ 默认平台是 `Windows,Android`，无需显式传 `--platforms`；只有要覆盖默认平台时才传。`--interval` 是正常检查周期，默认 `1h`；watch/daemon 模式还会在每天北京时间（UTC+8）`03:00`、`16:00`、`18:00` 强制执行一次自动刷新，该轮会注入 `force=true`，并且会中断普通 interval 的 sleep。`--error-retry` 是下载、发现或校验失败后的重试周期，默认 `60s`，也可以用 `--error-retry-seconds 60`。CLI 默认启动时向 stderr 打印 `BlueArchiveToolkit` ASCII banner，并把阶段进度日志写到 stderr，包括自动发现、proxy、server-info、marker、catalog、audit、download、snapshot 和 publish 阶段；download 阶段会输出已完成计数和单文件开始/完成状态，worker 从共享队列独立领取任务并在完成后立即领取下一项，完成计数保持单调不倒退，最终 report 的 `items` 仍按 pull plan 顺序排列，audit 阶段会输出官方 `.hash`、本地 BLAKE3、需修复项和 ZIP 结构校验结果摘要。daemon 还会写 `bat-events.jsonl` 结构化日志并按大小轮转。命令结果默认以人类可读摘要写到 stdout。需要纯机器输出时加 `--json --no-progress`，需要显式开启进度日志则用 `--progress`；只想关闭横幅但保留日志时可加 `--no-banner`。错误时 stderr 输出 JSON error，watch 模式下错误 JSON 的 `next_retry_seconds` 使用失败重试周期；如果未关闭 progress，错误 JSON 前可能已有 banner 和进度日志。普通错误 exit `1`，资源目录锁冲突 exit `75`，`verify` 或 `doctor` 发现问题也返回非 0。 -生产可以直接运行 `--watch`，也可以用 `--daemon` 后台运行，或者用 systemd service、容器或 Go 进程守护它。cron/systemd timer 仍可调用单次模式，但不再是 Rust 自动更新的唯一方式。下载默认并发 8，可用 `--download-concurrency` / `BAT_DOWNLOAD_CONCURRENCY` 配置为 `1..=256`；worker 动态领取共享 plan，finished 进度即时按完成数统计，发布 report 仍按 plan 顺序。项目是否热更新、热重载或重启进程，由上层业务集成决定。生产官方资源目录应使用独立输出目录，不要指向现有客户端或人工维护的资源目录；上层读取原版资源时应读取 `--output/current`，不要读取 `.staging` 或 `versions` 中未切换的目录。汉化 Patch/导出应写入 `--localized-output`，并保留官方相对目录结构，不能写回 `--output/current`。发布状态分两档：`not_localized` 只发布原版资源、不发布汉化资源；`localized` 发布原版和汉化两套资源。非 dry-run 每轮会创建 `--output/.official-sync.lock`，防止并发写同一官方资源目录；live daemon 还会阻止前台写命令直接修改它正在管理的同一目录。 +生产可以直接运行 `--watch`，也可以用 `--daemon` 后台运行，或者用 systemd service、容器或 Go 进程守护它。cron/systemd timer 仍可调用单次模式，但不再是 Rust 自动更新的唯一方式。下载默认并发 8，可用 `--download-concurrency` / `BAT_DOWNLOAD_CONCURRENCY` 配置为 `1..=256`；worker 动态领取共享 plan，finished 进度即时按完成数统计，发布 report 仍按 plan 顺序。项目是否热更新、热重载或重启进程，由上层业务集成决定。生产官方资源目录应使用独立输出目录，不要指向现有客户端或人工维护的资源目录；上层读取原版资源时应读取 `--output/current`，不要读取 `.staging` 或 `versions` 中未切换的目录。汉化 Patch/导出应写入 `--localized-output`，并保留官方相对目录结构，不能写回 `--output/current`。发布状态分两档：`not_localized` 只发布原版资源、不发布汉化资源；`localized` 发布原版和汉化两套资源。非 dry-run 每轮会创建 `--output/.official-sync.lock`，防止并发写同一官方资源目录；`release.cleanup` execute 使用同一个锁并在锁内重新生成/校验 `plan_id`，localized cleanup 使用 `.localized-release.lock`；live daemon 还会阻止前台写命令直接修改它正在管理的同一目录。 需要只做探测时可以加 `--dry-run`。需要关闭本地 audit 或 repair 时可以显式使用 `--no-audit-local` 或 `--no-repair`，但生产同步默认应保持开启。 diff --git a/docs/reference/rpc-backend-api.md b/docs/reference/rpc-backend-api.md index 55caabb..4354a64 100644 --- a/docs/reference/rpc-backend-api.md +++ b/docs/reference/rpc-backend-api.md @@ -178,7 +178,7 @@ 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": "...", "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.distribution` | 已实现 | `{ "channel": "official", "release_id": "...", "destination": "...", "offset": 0, "limit": 1000 }`，均可省略 | Rust 选择的 verified `resource_root` 和 download manifest entries；有 `destination` 时是 single-entry lookup，响应固定 `total=1, offset=0, limit=1, entries.length=1`，只校验该实际文件的 bytes/BLAKE3；无 `destination` 时保留管理查询分页语义。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 diff --git a/docs/reports/CURRENT_GAPS.md b/docs/reports/CURRENT_GAPS.md index 4fbeb80..59cf9a5 100644 --- a/docs/reports/CURRENT_GAPS.md +++ b/docs/reports/CURRENT_GAPS.md @@ -126,11 +126,15 @@ rollback 与 cleanup 保持独立；缺少 generic manifest 的旧 localized rel 明确标记 `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 +release-local CAS 引用通过持久化 `ownership_id + ordinal` ledger 幂等释放；没有 +`ownership_id` 的旧清单走明确 legacy cleanup path，不随机迁移 ownership。官方历史复用 +只对不可变文件使用 hard link，`translation-tasks.sqlite` 及 WAL/SHM 始终独立复制； +localized output 使用单写者锁和事务日志恢复 publish/rollback，publish 只有最终 +`verified` phase 才能 roll-forward，并在发布时写入实际 localized bytes/BLAKE3 的 +distribution manifest。`release.distribution(destination=...)` 是单条 lookup，返回 +exactly one entry；分发读取只使用轻量发布 metadata，保留 path ownership、symlink 和 +文件完整性检查；`release.cleanup execute` 与 official sync 共用 `.official-sync.lock`。 +P2 尚未由本轮处理：ResourceRepository 更完整的查询/权限/损坏恢复、模糊 TM、bat.sock peer credential/perms、FFI 生命周期、 资源大小/限额与更强的持久化 fsync 语义仍按后续专项推进。 diff --git a/infrastructure/src/cas/filesystem.rs b/infrastructure/src/cas/filesystem.rs index 627c431..8c63ae7 100644 --- a/infrastructure/src/cas/filesystem.rs +++ b/infrastructure/src/cas/filesystem.rs @@ -33,14 +33,14 @@ impl FileSystemCasRepository { /// Releases one release-owned CAS reference exactly once. pub async fn release_reference_once( &self, - release_id: &str, + ownership_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) + .release_reference_once(ownership_id, ordinal, &hash) .await .map_err(Self::map_error) } diff --git a/infrastructure/src/localized_patch.rs b/infrastructure/src/localized_patch.rs index 2586870..6af8a93 100644 --- a/infrastructure/src/localized_patch.rs +++ b/infrastructure/src/localized_patch.rs @@ -100,6 +100,8 @@ struct LocalizedReleaseTransaction { current_target: Option, previous_state_bytes: Option>, new_state: Option, + #[serde(default)] + rollback_backup_path: Option, } impl LocalizedReleaseTransaction { @@ -121,6 +123,7 @@ impl LocalizedReleaseTransaction { current_target: Some(Path::new(LOCALIZED_VERSIONS_DIR).join(release_id)), previous_state_bytes, new_state: None, + rollback_backup_path: None, } } } @@ -941,9 +944,33 @@ impl LocalizedPatchService { current_target: manifest.rollback.previous_current_target.clone(), previous_state_bytes: Some(previous_state_bytes), new_state: Some(new_state.clone()), + rollback_backup_path: Some(localized_output_root.join(".rollback").join(format!( + "{}.{}", + current_release_id, + std::process::id() + ))), }; write_localized_transaction(localized_output_root, &transaction)?; let mutation_result = (|| -> anyhow::Result<()> { + let backup_path = transaction + .rollback_backup_path + .as_deref() + .ok_or_else(|| anyhow::anyhow!("localized rollback 缺少备份路径"))?; + ensure_path_within_root(localized_output_root, backup_path) + .map_err(anyhow::Error::msg)?; + ensure_safe_directory_path( + backup_path.parent().unwrap_or(localized_output_root), + "localized rollback 备份目录", + ) + .map_err(anyhow::Error::msg)?; + fs::create_dir_all(backup_path.parent().unwrap_or(localized_output_root))?; + ensure_safe_directory_path( + backup_path.parent().unwrap_or(localized_output_root), + "localized rollback 备份目录", + ) + .map_err(anyhow::Error::msg)?; + fs::rename(&remove_version_path, backup_path)?; + update_localized_transaction_phase(localized_output_root, "version_staged")?; restore_current_symlink( localized_output_root, ¤t_path, @@ -952,8 +979,18 @@ impl LocalizedPatchService { 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)?; + if let Some(previous_target) = manifest.rollback.previous_current_target.as_deref() { + if !current_points_to_version( + ¤t_path, + &localized_output_root.join(previous_target), + )? { + return Err(anyhow::anyhow!("localized rollback current 最终校验失败")); + } + } else if fs::symlink_metadata(¤t_path).is_ok() { + return Err(anyhow::anyhow!("localized rollback 应移除 current 指针")); + } update_localized_transaction_phase(localized_output_root, "version_removed")?; + remove_owned_path(backup_path)?; Ok(()) })(); if let Err(error) = mutation_result { @@ -1216,6 +1253,7 @@ impl LocalizedPatchService { translation_workflow_status, updated_unix_seconds: unix_seconds_now(), }; + update_localized_transaction_state(&config.localized_output_root, &state)?; write_file_atomic( &state_path, &serde_json::to_vec_pretty(&state)?, @@ -1230,6 +1268,7 @@ impl LocalizedPatchService { ¤t_path, &config.unzip_command, )?; + update_localized_transaction_phase(&config.localized_output_root, "verified")?; Ok(LocalizedPatchReport { version_path, @@ -2139,6 +2178,7 @@ fn verify_published_localized_release( &manifest, unzip_command, )?; + verify_localized_distribution_manifest_at(version_path, &manifest.localized_release_id)?; integrity.current_points_to_release = current_points_to_version(current_path, version_path)?; if !integrity.current_points_to_release { return Err(anyhow::anyhow!( @@ -2147,6 +2187,23 @@ fn verify_published_localized_release( version_path.display() )); } + let state_path = current_path + .parent() + .ok_or_else(|| anyhow::anyhow!("localized current 缺少输出根目录"))? + .join(LOCALIZED_VERSION_STATE_FILE); + let state = read_localized_version_state( + current_path + .parent() + .ok_or_else(|| anyhow::anyhow!("localized current 缺少输出根目录"))?, + )? + .ok_or_else(|| anyhow::anyhow!("缺少 localized version state：{}", state_path.display()))?; + if state.current_release_id.as_deref() != Some(manifest.localized_release_id.as_str()) { + return Err(anyhow::anyhow!( + "localized version state 与发布 manifest 不一致：expected={} actual={:?}", + manifest.localized_release_id, + state.current_release_id + )); + } Ok(integrity) } @@ -2289,6 +2346,24 @@ fn update_localized_transaction_phase( write_localized_transaction(localized_output_root, &transaction) } +fn update_localized_transaction_state( + localized_output_root: &Path, + state: &LocalizedVersionState, +) -> 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.new_state = Some(state.clone()); + 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) { @@ -2341,15 +2416,14 @@ fn recover_localized_transaction(localized_output_root: &Path) -> anyhow::Result == Some(expected) }); let publish_committed = transaction.operation == "publish" + && transaction.phase == "verified" && 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; + && state_matches; + let rollback_committed = transaction.operation == "rollback" + && transaction.phase == "version_removed" + && current_matches + && state_matches; if publish_committed { if let Some(staging) = transaction.staging_path.as_deref() { @@ -2357,6 +2431,9 @@ fn recover_localized_transaction(localized_output_root: &Path) -> anyhow::Result } } else if rollback_committed { remove_owned_path(&transaction.version_path)?; + if let Some(backup) = transaction.rollback_backup_path.as_deref() { + remove_owned_path(backup)?; + } } else { if let Some(target) = transaction.previous_current_target.as_deref() { let target_path = localized_output_root.join(target); @@ -2372,6 +2449,14 @@ fn recover_localized_transaction(localized_output_root: &Path) -> anyhow::Result } remove_owned_path(&transaction.version_path)?; } + if transaction.operation == "rollback" { + if let Some(backup) = transaction.rollback_backup_path.as_deref() { + if fs::symlink_metadata(backup).is_ok() { + remove_owned_path(&transaction.version_path)?; + fs::rename(backup, &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), @@ -3533,6 +3618,98 @@ mod tests { assert!(!localized_output_transaction_pending(&root).unwrap()); } + #[cfg(unix)] + #[test] + fn publish_recovery_requires_verified_phase_before_roll_forward() { + use std::os::unix::fs::symlink; + + for (phase, new_is_current, should_keep_new) in [ + ("prepared", false, false), + ("version_published", false, false), + ("current_switched", true, false), + ("state_written", true, false), + ("verification_failed", true, false), + ("verification_succeeded", true, false), + ("verified", true, true), + ] { + let temp = TempDir::new().unwrap(); + let root = temp.path().join("localized"); + let old = root.join(LOCALIZED_VERSIONS_DIR).join("old"); + let new = root.join(LOCALIZED_VERSIONS_DIR).join("new"); + let staging = root.join(LOCALIZED_STAGING_DIR).join("new"); + fs::create_dir_all(&old).unwrap(); + fs::create_dir_all(&new).unwrap(); + fs::create_dir_all(&staging).unwrap(); + symlink( + Path::new(LOCALIZED_VERSIONS_DIR).join(if new_is_current { "new" } else { "old" }), + root.join(LOCALIZED_CURRENT_LINK), + ) + .unwrap(); + let old_state = LocalizedVersionState { + state_version: LOCALIZED_VERSION_STATE_VERSION, + official_release_id: "official-old".to_string(), + current_release_id: Some("old".to_string()), + status: "localized".to_string(), + translation_workflow_status: None, + updated_unix_seconds: 1, + }; + let new_state = LocalizedVersionState { + current_release_id: Some("new".to_string()), + updated_unix_seconds: 2, + ..old_state.clone() + }; + let old_state_bytes = serde_json::to_vec_pretty(&old_state).unwrap(); + let state_bytes = if new_is_current { + serde_json::to_vec_pretty(&new_state).unwrap() + } else { + old_state_bytes.clone() + }; + write_file_atomic( + &root.join(LOCALIZED_VERSION_STATE_FILE), + &state_bytes, + STATE_FILE_MODE, + "test state", + ) + .unwrap(); + let mut transaction = LocalizedReleaseTransaction::publish( + "new", + new.clone(), + staging.clone(), + Some(PathBuf::from("versions/old")), + Some(old_state_bytes), + ); + transaction.phase = phase.to_string(); + transaction.new_state = Some(new_state.clone()); + write_localized_transaction(&root, &transaction).unwrap(); + + recover_localized_transaction(&root).unwrap(); + + if should_keep_new { + assert_eq!( + fs::read_link(root.join(LOCALIZED_CURRENT_LINK)).unwrap(), + PathBuf::from("versions/new") + ); + assert_eq!( + read_localized_version_state(&root).unwrap(), + Some(new_state) + ); + assert!(new.exists()); + } else { + assert_eq!( + fs::read_link(root.join(LOCALIZED_CURRENT_LINK)).unwrap(), + PathBuf::from("versions/old") + ); + assert_eq!( + read_localized_version_state(&root).unwrap(), + Some(old_state) + ); + assert!(!new.exists()); + } + assert!(!staging.exists()); + assert!(!localized_output_transaction_pending(&root).unwrap()); + } + } + fn push_i16_le(data: &mut Vec, value: i16) { data.extend_from_slice(&value.to_le_bytes()); } diff --git a/infrastructure/src/official_download.rs b/infrastructure/src/official_download.rs index dc9997d..6229228 100644 --- a/infrastructure/src/official_download.rs +++ b/infrastructure/src/official_download.rs @@ -25,6 +25,7 @@ use std::fs::{self, File}; use std::io::Read; use std::path::{Path, PathBuf}; use std::process::Command; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -74,6 +75,23 @@ const OFFICIAL_CAS_REUSE_REFERENCES_VERSION: u32 = 1; const DOWNLOAD_MANIFEST_VERSION: u32 = 1; const DOWNLOAD_QUARANTINE_VERSION: u32 = 1; const DEFAULT_RETRY_ATTEMPTS: usize = 3; +static CAS_OWNERSHIP_SEQUENCE: AtomicU64 = AtomicU64::new(1); + +fn new_cas_ownership_id(output_root: &Path) -> String { + let sequence = CAS_OWNERSHIP_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_nanos()) + .unwrap_or_default(); + let material = format!( + "{}:{}:{}:{}", + output_root.display(), + std::process::id(), + now, + sequence + ); + format!("cas-owner-{}", blake3::hash(material.as_bytes()).to_hex()) +} /// Outcome for one official resource pull item. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -353,6 +371,12 @@ pub struct OfficialResourceReuseWarning { pub struct OfficialCasReuseReferenceManifest { /// 引用清单版本。 pub version: u32, + /// Persistent ownership identity for this release generation. + /// + /// `None` is an explicit legacy manifest marker. Legacy cleanup keeps the + /// historical basename key and never invents a new identity while loading. + #[serde(default)] + pub ownership_id: Option, /// 每个 CAS 引用一项。允许重复，因为每个拉取项分别拥有一个引用。 pub object_ids: Vec, } @@ -361,6 +385,7 @@ impl Default for OfficialCasReuseReferenceManifest { fn default() -> Self { Self { version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION, + ownership_id: None, object_ids: Vec::new(), } } @@ -393,15 +418,17 @@ pub fn read_cas_reuse_reference_manifest_at( /// Decrements and removes CAS references recorded for a release. /// -/// 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. +/// Each decrement is committed together with a durable `(ownership, ordinal)` +/// 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. Legacy manifests without `ownership_id` intentionally use +/// the historical release-basename key and are never assigned a new identity +/// during cleanup. 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 + let legacy_release_id = release_root .file_name() .and_then(|name| name.to_str()) .filter(|name| !name.is_empty() && *name != "." && *name != "..") @@ -412,6 +439,7 @@ pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Res ) })? .to_string(); + let ownership_id = manifest.ownership_id.clone().unwrap_or(legacy_release_id); let objects_root = cas_root.join("objects"); let metadata_path = cas_root.join("metadata.sqlite"); require_existing_directory(cas_root, "CAS 根目录")?; @@ -422,14 +450,14 @@ pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Res 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 ownership_id_for_runtime = ownership_id.clone(); let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build() .map_err(|error| format!("创建 CAS 引用清理 runtime 失败：{error}"))?; let did_release = runtime.block_on(async move { let cas = crate::FileSystemCasRepository::new(cas_root); - cas.release_reference_once(&release_id_for_runtime, ordinal, &object_id_for_runtime) + cas.release_reference_once(&ownership_id_for_runtime, ordinal, &object_id_for_runtime) .await .map_err(|error| format!("减少 CAS release 引用失败 object={object_id}：{error}")) })?; @@ -857,8 +885,14 @@ impl OfficialResourcePullService { if object_ids.is_empty() { return Ok(()); } - let mut manifest = - read_cas_reuse_reference_manifest_at(&self.output_root)?.unwrap_or_default(); + let mut manifest = match read_cas_reuse_reference_manifest_at(&self.output_root)? { + Some(manifest) => manifest, + None => OfficialCasReuseReferenceManifest { + version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION, + ownership_id: Some(new_cas_ownership_id(&self.output_root)), + object_ids: Vec::new(), + }, + }; manifest.object_ids.extend(object_ids); write_cas_reuse_reference_manifest(&self.output_root, &manifest) } @@ -4662,7 +4696,9 @@ exit 22 let references = read_cas_reuse_reference_manifest_at(&out_dir) .unwrap() .unwrap(); + assert!(references.ownership_id.is_some()); assert_eq!(references.object_ids, vec![object_id.clone()]); + let ownership_id = references.ownership_id.clone().unwrap(); assert_eq!(cas_reference_count(&cas_root, &object_id), 2); assert_eq!( @@ -4680,6 +4716,7 @@ exit 22 out_dir.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE), serde_json::to_vec(&OfficialCasReuseReferenceManifest { version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION, + ownership_id: Some(ownership_id), object_ids: vec![object_id.clone()], }) .unwrap(), @@ -4695,6 +4732,75 @@ exit 22 .is_none()); } + #[test] + fn cas_ownership_identity_isolated_for_same_release_basename() { + let temp = TempDir::new().unwrap(); + let cas_root = temp.path().join("cas"); + let object_id = store_cas_object(&cas_root, b"shared object"); + let cas = crate::FileSystemCasRepository::new(&cas_root); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + cas.add_reference(&object_id).await.unwrap(); + cas.add_reference(&object_id).await.unwrap(); + }); + + let first = temp.path().join("first").join("release"); + let second = temp.path().join("second").join("release"); + fs::create_dir_all(&first).unwrap(); + fs::create_dir_all(&second).unwrap(); + for (root, ownership_id) in [(&first, "owner-first"), (&second, "owner-second")] { + fs::write( + root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE), + serde_json::to_vec(&OfficialCasReuseReferenceManifest { + version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION, + ownership_id: Some(ownership_id.to_string()), + object_ids: vec![object_id.clone()], + }) + .unwrap(), + ) + .unwrap(); + } + + assert_eq!(release_cas_reuse_references(&first, &cas_root).unwrap(), 1); + assert_eq!(cas_reference_count(&cas_root, &object_id), 2); + assert_eq!(release_cas_reuse_references(&second, &cas_root).unwrap(), 1); + assert_eq!(cas_reference_count(&cas_root, &object_id), 1); + } + + #[test] + fn legacy_cas_manifest_uses_basename_key_without_migration() { + let temp = TempDir::new().unwrap(); + let cas_root = temp.path().join("cas"); + let object_id = store_cas_object(&cas_root, b"legacy object"); + let release_root = temp.path().join("legacy-release"); + fs::create_dir_all(&release_root).unwrap(); + fs::write( + release_root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE), + serde_json::to_vec(&OfficialCasReuseReferenceManifest { + version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION, + ownership_id: None, + object_ids: vec![object_id.clone()], + }) + .unwrap(), + ) + .unwrap(); + + let cas = crate::FileSystemCasRepository::new(&cas_root); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { cas.add_reference(&object_id).await.unwrap() }); + assert_eq!( + release_cas_reuse_references(&release_root, &cas_root).unwrap(), + 1 + ); + assert_eq!(cas_reference_count(&cas_root, &object_id), 1); + } + #[test] fn corrupted_cas_falls_back_to_network_with_diagnostic() { let temp = TempDir::new().unwrap(); diff --git a/infrastructure/src/official_update.rs b/infrastructure/src/official_update.rs index 8c40641..56bec3f 100644 --- a/infrastructure/src/official_update.rs +++ b/infrastructure/src/official_update.rs @@ -13,6 +13,7 @@ use crate::official_changes::{ write_official_resource_change_handoff, OfficialResourceChangeHandoffReport, OfficialResourceChangeSummary, }; +use crate::official_download::OFFICIAL_CAS_REUSE_REFERENCES_FILE; use crate::official_game_main_config::{ resolve_game_main_config_source, OfficialGameMainConfigSelectedSource, OfficialGameMainConfigSourceKind, @@ -1128,7 +1129,20 @@ impl OfficialPublishLayout { if !path_exists_no_follow(active_root)? { return Ok(()); } - copy_tree_no_symlink(active_root, staging_root, active_root == self.root) + copy_tree_no_symlink(active_root, staging_root, active_root == self.root)?; + if active_root != self.root { + let cas_references = staging_root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE); + if path_exists_no_follow(&cas_references)? { + ensure_safe_file_target(staging_root, &cas_references, "staging CAS 引用清单")?; + fs::remove_file(&cas_references).map_err(|error| { + format!( + "清理 active release CAS ownership 清单失败 {}：{error}", + cas_references.display() + ) + })?; + } + } + Ok(()) } fn legacy_manifest_exists(&self) -> Result { @@ -4309,21 +4323,22 @@ fn required_platform(endpoint: &YostarJpResourceEndpoint) -> anyhow::Result anyhow::Result { - validate_output_root(&config.output_root).map_err(anyhow::Error::msg)?; - ensure_safe_directory_path(&config.output_root, "资源输出目录") - .map_err(anyhow::Error::msg)?; - fs::create_dir_all(&config.output_root)?; - ensure_safe_directory_path(&config.output_root, "资源输出目录") - .map_err(anyhow::Error::msg)?; - let path = config.lock_path(); - ensure_safe_file_target(&config.output_root, &path, "官方同步锁") - .map_err(anyhow::Error::msg)?; + Self::acquire_output_root(&config.output_root) + } + + pub(crate) fn acquire_output_root(output_root: &Path) -> anyhow::Result { + validate_output_root(output_root).map_err(anyhow::Error::msg)?; + ensure_safe_directory_path(output_root, "资源输出目录").map_err(anyhow::Error::msg)?; + fs::create_dir_all(output_root)?; + ensure_safe_directory_path(output_root, "资源输出目录").map_err(anyhow::Error::msg)?; + let path = output_root.join(".official-sync.lock"); + ensure_safe_file_target(output_root, &path, "官方同步锁").map_err(anyhow::Error::msg)?; for attempt in 0..=1 { let mut options = OpenOptions::new(); options.write(true).create_new(true); @@ -4361,6 +4376,12 @@ impl OfficialUpdateLock { } } +pub(crate) fn acquire_official_output_lock( + output_root: &Path, +) -> anyhow::Result { + OfficialUpdateLock::acquire_output_root(output_root) +} + impl Drop for OfficialUpdateLock { fn drop(&mut self) { let expected = std::process::id().to_string(); diff --git a/infrastructure/src/release_ops.rs b/infrastructure/src/release_ops.rs index fc89929..4e009dc 100644 --- a/infrastructure/src/release_ops.rs +++ b/infrastructure/src/release_ops.rs @@ -6,12 +6,13 @@ use crate::localized_patch::{ inspect_localized_release_artifact_at, read_localized_patch_manifest_at, - read_localized_version_state, LocalizedDistributionManifest, LOCALIZED_CURRENT_LINK, - LOCALIZED_DISTRIBUTION_MANIFEST_FILE, LOCALIZED_STAGING_DIR, LOCALIZED_VERSIONS_DIR, + read_localized_version_state, LocalizedDistributionEntry, 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, - OfficialDownloadManifest, + OfficialDownloadManifest, OfficialDownloadManifestEntry, }; use crate::official_update::{read_version_state, OfficialVersionRecord, OfficialVersionState}; use crate::path_security::{ @@ -363,15 +364,25 @@ pub fn select_release_distribution( "请求的 release 不存在、publication identity 无效或不是当前 release", )); }; + let single_entry_lookup = params.destination.is_some(); let all_entries = selection.entries; - let total = all_entries.len(); - let offset = params.offset.min(total); - let limit = if params.limit == 0 { - 1000 + let (total, offset, limit, entries) = if single_entry_lookup { + (1, 0, 1, all_entries) } else { - params.limit.min(1000) + let total = all_entries.len(); + let offset = params.offset.min(total); + let limit = if params.limit == 0 { + 1000 + } else { + params.limit.min(1000) + }; + ( + total, + offset, + limit, + all_entries.into_iter().skip(offset).take(limit).collect(), + ) }; - let entries = all_entries.into_iter().skip(offset).take(limit).collect(); Ok(ReleaseDistributionPage { available: true, channel: channel.as_str().to_string(), @@ -455,16 +466,35 @@ fn select_release_distribution_metadata( 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 { + let entries = if let Some(destination) = destination { + let mut matching_entries = manifest + .entries + .values() + .filter(|entry| entry.destination == destination); + let Some(entry) = matching_entries.next() else { + return Ok(None); + }; + if matching_entries.next().is_some() { + return Ok(None); + } + vec![ReleaseDistributionEntry { url: entry.url.clone(), destination: entry.destination.clone(), bytes: entry.bytes, blake3: entry.blake3.clone(), - }) - .collect::>(); + }] + } else { + 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); } @@ -502,19 +532,51 @@ fn select_release_distribution_metadata( }) 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, + let entries = if let Some(destination) = destination { + let mut matching_entries = manifest + .entries + .iter() + .filter(|entry| entry.destination == destination); + let Some(entry) = matching_entries.next() else { + return Ok(None); + }; + if matching_entries.next().is_some() { + return Ok(None); + } + let mut official_entries = official_manifest + .entries + .values() + .filter(|official_entry| official_entry.destination == entry.destination); + let Some(official_entry) = official_entries.next() else { + return Ok(None); + }; + if official_entries.next().is_some() { + return Ok(None); + } + if !localized_distribution_entry_matches_official(entry, official_entry) { + return Ok(None); + } + vec![ReleaseDistributionEntry { + url: entry.url.clone(), + destination: entry.destination.clone(), bytes: entry.bytes, - blake3: entry.blake3, - }) - .collect::>(); + blake3: entry.blake3.clone(), + }] + } else { + if !localized_distribution_matches_official(&manifest, &official_manifest) { + return Ok(None); + } + 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); } @@ -534,38 +596,44 @@ fn validate_distribution_entries( entries: &[ReleaseDistributionEntry], destination: Option<&str>, ) -> anyhow::Result { + if let Some(destination) = destination { + let Some(entry) = entries + .iter() + .find(|entry| entry.destination == destination) + else { + return Ok(false); + }; + let path = distribution_file_path(root, &entry.destination)?; + let bytes = fs::read(&path)?; + return Ok(bytes.len() as u64 == entry.bytes + && blake3::hash(&bytes).to_hex().to_string() == entry.blake3); + } 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 path = distribution_file_path(root, &entry.destination)?; + 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); - }; + Ok(true) +} + +fn distribution_file_path(root: &Path, destination: &str) -> anyhow::Result { + if destination.is_empty() || destination.contains('\\') || destination.contains('\0') { + return Err(anyhow::anyhow!( + "release distribution destination 不安全：{destination}" + )); + } let path = root.join(destination); + ensure_path_within_root(root, &path).map_err(anyhow::Error::msg)?; 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) + Ok(path) } fn localized_distribution_matches_official( @@ -589,6 +657,13 @@ fn localized_distribution_matches_official( }) } +fn localized_distribution_entry_matches_official( + localized: &LocalizedDistributionEntry, + official: &OfficialDownloadManifestEntry, +) -> bool { + localized.destination == official.destination && localized.url == official.url +} + fn blocked_distribution( channel: Channel, release_id: Option, @@ -624,8 +699,23 @@ pub fn cleanup_releases( params: &ReleaseCleanupParams, unzip_command: &Path, ) -> anyhow::Result { - let _localized_lock = crate::localized_patch::acquire_localized_output_lock(localized_root)?; - crate::localized_patch::recover_localized_output_transaction(localized_root)?; + let _official_lock = if params.execute { + Some(crate::official_update::acquire_official_output_lock( + official_root, + )?) + } else { + None + }; + let _localized_lock = if params.execute { + Some(crate::localized_patch::acquire_localized_output_lock( + localized_root, + )?) + } else { + None + }; + if params.execute { + crate::localized_patch::recover_localized_output_transaction(localized_root)?; + } let plan = build_cleanup_plan(official_root, localized_root, cas_root, unzip_command)?; if !params.execute { return Ok(ReleaseCleanupReport { @@ -1566,6 +1656,76 @@ mod tests { assert!(selected.resource_root.is_none()); } + #[test] + fn distribution_destination_uses_one_entry_without_hashing_other_entries() { + let temp = tempfile::tempdir().unwrap(); + let official_root = temp.path().join("official"); + let version = official_root.join(OFFICIAL_VERSIONS_DIR).join("large"); + fs::create_dir_all(&version).unwrap(); + let mut entries = BTreeMap::new(); + let target = "resource-0000.bin"; + for index in 0..5000 { + let destination = format!("resource-{index:04}.bin"); + let data = format!("resource-{index}").into_bytes(); + fs::write(version.join(&destination), &data).unwrap(); + entries.insert( + format!("https://example.invalid/{destination}"), + OfficialDownloadManifestEntry { + url: format!("https://example.invalid/{destination}"), + destination, + bytes: data.len() as u64, + blake3: if index == 0 { + blake3::hash(&data).to_hex().to_string() + } else { + "0".repeat(64) + }, + }, + ); + } + fs::write( + version.join("official-download-manifest.json"), + serde_json::to_vec(&OfficialDownloadManifest { + version: 1, + entries, + }) + .unwrap(), + ) + .unwrap(); + fs::create_dir_all(&official_root).unwrap(); + symlink( + Path::new(OFFICIAL_VERSIONS_DIR).join("large"), + official_root.join(OFFICIAL_CURRENT_LINK), + ) + .unwrap(); + fs::write( + official_root.join(OFFICIAL_VERSION_STATE_FILE), + serde_json::to_vec(&OfficialVersionState { + current_completed_version: Some(official_record(&official_root, "large")), + ..OfficialVersionState::default() + }) + .unwrap(), + ) + .unwrap(); + + let selected = select_release_distribution( + &official_root, + &temp.path().join("localized"), + &ReleaseDistributionParams { + channel: Some("official".to_string()), + destination: Some(target.to_string()), + ..ReleaseDistributionParams::default() + }, + Path::new("unzip"), + ) + .unwrap(); + assert!(selected.available); + assert_eq!(selected.total, 1); + assert_eq!(selected.offset, 0); + assert_eq!(selected.limit, 1); + assert_eq!(selected.entries.len(), 1); + assert_eq!(selected.entries[0].destination, target); + } + #[test] fn cleanup_plan_protects_current_and_removes_only_unreferenced_history() { let temp = tempfile::tempdir().unwrap(); @@ -1676,6 +1836,30 @@ mod tests { .exists()); } + #[test] + fn cleanup_execute_uses_the_official_sync_filesystem_lock() { + let temp = tempfile::tempdir().unwrap(); + let official_root = temp.path().join("official"); + let localized_root = temp.path().join("localized"); + fs::create_dir_all(&official_root).unwrap(); + let sync_lock = + crate::official_update::acquire_official_output_lock(&official_root).unwrap(); + + let result = cleanup_releases( + &official_root, + &localized_root, + &temp.path().join("cas"), + &ReleaseCleanupParams { + execute: true, + plan_id: Some("unused".to_string()), + }, + Path::new("unzip"), + ); + + assert!(result.unwrap_err().to_string().contains("锁定")); + drop(sync_lock); + } + #[cfg(unix)] #[test] fn cleanup_retains_symlinks_and_localized_identity_injection() { diff --git a/internal/api/openapi.go b/internal/api/openapi.go index 8f15182..fb36636 100644 --- a/internal/api/openapi.go +++ b/internal/api/openapi.go @@ -109,7 +109,7 @@ paths: type: string - name: destination in: query - description: Optional release-relative path whose localized bytes and BLAKE3 are revalidated. + description: Optional release-relative path for single-entry lookup; Rust returns exactly one entry and revalidates the selected channel's actual bytes and BLAKE3. schema: type: string - name: offset diff --git a/internal/api/release_http.go b/internal/api/release_http.go index 73af42f..4a14c44 100644 --- a/internal/api/release_http.go +++ b/internal/api/release_http.go @@ -194,9 +194,18 @@ func (s *Server) loadReleaseDistributionFrom(r *http.Request, params backendrpc. } pageSize := 1000 result, err := backend.ReleaseDistribution(r.Context(), params) - if err != nil || result == nil || !result.Available || result.Total <= len(result.Entries) { + if err != nil || result == nil || !result.Available { return result, err } + if params.Destination != "" { + if result.Total != 1 || result.Offset != 0 || result.Limit != 1 || len(result.Entries) != 1 { + return nil, &releaseSelectorError{message: "Rust single-entry release distribution response is invalid"} + } + return result, nil + } + if result.Total <= len(result.Entries) { + return result, nil + } all := append([]backendrpc.ReleaseDistributionEntry(nil), result.Entries...) for offset := len(all); offset < result.Total; { next, nextErr := backend.ReleaseDistribution(r.Context(), backendrpc.ReleaseDistributionParams{ diff --git a/internal/api/release_http_test.go b/internal/api/release_http_test.go index eeba4d4..8df4774 100644 --- a/internal/api/release_http_test.go +++ b/internal/api/release_http_test.go @@ -4,6 +4,9 @@ import ( "context" "net/http" "net/http/httptest" + "os" + "path/filepath" + "strconv" "strings" "testing" @@ -15,6 +18,49 @@ type releaseBackendStub struct { root string available bool distributionParams []backendrpc.ReleaseDistributionParams + largeDistribution bool +} + +type variableDistributionBackend struct { + *releaseBackendStub + officialRoot string + localizedRoot string + officialBytes []byte + localizedBytes []byte +} + +func (b *variableDistributionBackend) ReleaseDistribution(_ context.Context, params backendrpc.ReleaseDistributionParams) (*backendrpc.ReleaseDistributionPage, error) { + b.distributionParams = append(b.distributionParams, params) + root := b.officialRoot + data := b.officialBytes + channel := "official" + releaseID := "official-1" + hash := "official-b3" + if params.Channel == "localized" { + root = b.localizedRoot + data = b.localizedBytes + channel = "localized" + releaseID = "localized-1" + hash = "localized-b3" + } + return &backendrpc.ReleaseDistributionPage{ + Available: true, + Channel: channel, + ReleaseID: releaseID, + ResourceRoot: root, + Status: "ready", + StatusCode: "distribution.ready", + ArtifactIntegrityStatus: "valid", + Total: 1, + Offset: 0, + Limit: 1, + Entries: []backendrpc.ReleaseDistributionEntry{{ + URL: "https://prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes", + Destination: "prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes", + Bytes: uint64(len(data)), + BLAKE3: hash, + }}, + }, nil } func (b *controlBackend) ReleaseStatus(context.Context) (*backendrpc.ReleaseStatusReport, error) { @@ -63,6 +109,52 @@ func (b *releaseBackendStub) ReleaseList(context.Context, backendrpc.ReleaseList func (b *releaseBackendStub) ReleaseDistribution(_ context.Context, params backendrpc.ReleaseDistributionParams) (*backendrpc.ReleaseDistributionPage, error) { b.distributionParams = append(b.distributionParams, params) + if b.largeDistribution { + target := "prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes" + if params.Destination != "" { + return &backendrpc.ReleaseDistributionPage{ + Available: b.available, + Channel: params.Channel, + ReleaseID: params.ReleaseID, + ResourceRoot: b.root, + Status: "ready", + StatusCode: "distribution.ready", + ArtifactIntegrityStatus: "valid", + Total: 1, + Offset: 0, + Limit: 1, + Entries: []backendrpc.ReleaseDistributionEntry{{ + URL: "https://prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes", + Destination: target, + Bytes: 21, + BLAKE3: "not-used-by-http-index", + }}, + }, nil + } + entries := make([]backendrpc.ReleaseDistributionEntry, 5000) + for index := range entries { + entries[index] = backendrpc.ReleaseDistributionEntry{ + URL: "https://prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/resource.bytes", + Destination: "prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/resource-" + strconv.Itoa(index) + ".bytes", + Bytes: 21, + BLAKE3: "not-used-by-http-index", + } + } + entries[0].Destination = target + return &backendrpc.ReleaseDistributionPage{ + Available: b.available, + Channel: params.Channel, + ReleaseID: params.ReleaseID, + ResourceRoot: b.root, + Status: "ready", + StatusCode: "distribution.ready", + ArtifactIntegrityStatus: "valid", + Total: len(entries), + Offset: 0, + Limit: len(entries), + Entries: entries, + }, nil + } return &backendrpc.ReleaseDistributionPage{ Available: b.available, Channel: params.Channel, @@ -72,7 +164,8 @@ func (b *releaseBackendStub) ReleaseDistribution(_ context.Context, params backe StatusCode: "distribution.ready", ArtifactIntegrityStatus: "valid", Total: 1, - Limit: 1000, + Offset: 0, + Limit: 1, Entries: []backendrpc.ReleaseDistributionEntry{{ URL: "https://prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes", Destination: "prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes", @@ -82,6 +175,113 @@ func (b *releaseBackendStub) ReleaseDistribution(_ context.Context, params backe }, nil } +func TestCDNSingleEntryLookupDoesNotPaginateLargeDistribution(t *testing.T) { + cfg := DefaultConfig() + cfg.RequireIndexed = false + if err := cfg.Normalize(); err != nil { + t.Fatal(err) + } + backend := &releaseBackendStub{ + fakeBackend: &fakeBackend{}, + root: fixtureRoot(t), + available: true, + largeDistribution: true, + } + server := NewServer(cfg, backend, nil) + request := httptest.NewRequest( + http.MethodGet, + "/prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes?channel=localized&release_id=localized-1", + nil, + ) + recorder := httptest.NewRecorder() + server.Handler().ServeHTTP(recorder, request) + if recorder.Code != http.StatusOK || recorder.Body.String() != "TABLE_CATALOG_FIXTURE" { + t.Fatalf("large distribution CDN status=%d body=%q", recorder.Code, recorder.Body.String()) + } + if len(backend.distributionParams) != 1 { + t.Fatalf("single-entry lookup made %d backend calls", len(backend.distributionParams)) + } + params := backend.distributionParams[0] + if params.Destination != "prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes" { + t.Fatalf("single-entry destination=%q", params.Destination) + } +} + +func TestCDNUsesLocalizedBytesAndHashForGetAndHead(t *testing.T) { + cfg := DefaultConfig() + cfg.RequireIndexed = false + if err := cfg.Normalize(); err != nil { + t.Fatal(err) + } + rel := filepath.FromSlash("prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes") + officialBytes := []byte("official-A") + localizedBytes := []byte("localized-B-with-a-different-length") + officialRoot := t.TempDir() + localizedRoot := t.TempDir() + if err := os.MkdirAll(filepath.Dir(filepath.Join(officialRoot, rel)), 0o755); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Dir(filepath.Join(localizedRoot, rel)), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(officialRoot, rel), officialBytes, 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(localizedRoot, rel), localizedBytes, 0o644); err != nil { + t.Fatal(err) + } + backend := &variableDistributionBackend{ + releaseBackendStub: &releaseBackendStub{fakeBackend: &fakeBackend{}}, + officialRoot: officialRoot, + localizedRoot: localizedRoot, + officialBytes: officialBytes, + localizedBytes: localizedBytes, + } + server := NewServer(cfg, backend, nil) + + localizedURL := "/prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes?channel=localized&release_id=localized-1" + recorder := httptest.NewRecorder() + server.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, localizedURL, nil)) + if recorder.Code != http.StatusOK || recorder.Body.String() != string(localizedBytes) { + t.Fatalf("localized GET status=%d body=%q", recorder.Code, recorder.Body.String()) + } + if recorder.Header().Get("ETag") != `"blake3-localized-b3"` { + t.Fatalf("localized ETag=%q", recorder.Header().Get("ETag")) + } + if recorder.Result().ContentLength != int64(len(localizedBytes)) { + t.Fatalf("localized Content-Length=%d", recorder.Result().ContentLength) + } + + recorder = httptest.NewRecorder() + server.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodHead, localizedURL, nil)) + if recorder.Code != http.StatusOK || recorder.Body.Len() != 0 { + t.Fatalf("localized HEAD status=%d body=%d", recorder.Code, recorder.Body.Len()) + } + if recorder.Header().Get("ETag") != `"blake3-localized-b3"` || + recorder.Result().ContentLength != int64(len(localizedBytes)) { + t.Fatalf("localized HEAD headers etag=%q length=%d", recorder.Header().Get("ETag"), recorder.Result().ContentLength) + } + + officialURL := "/prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes?channel=official&release_id=official-1" + recorder = httptest.NewRecorder() + server.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, officialURL, nil)) + if recorder.Code != http.StatusOK || recorder.Body.String() != string(officialBytes) { + t.Fatalf("official GET status=%d body=%q", recorder.Code, recorder.Body.String()) + } + if recorder.Header().Get("ETag") != `"blake3-official-b3"` || + recorder.Result().ContentLength != int64(len(officialBytes)) { + t.Fatalf("official headers etag=%q length=%d", recorder.Header().Get("ETag"), recorder.Result().ContentLength) + } + if len(backend.distributionParams) != 3 { + t.Fatalf("backend calls=%d want=3", len(backend.distributionParams)) + } + for _, params := range backend.distributionParams { + if params.Destination != "prod-clientpatch.bluearchiveyostar.com/r93_fixture/TableBundles/TableCatalog.bytes" { + t.Fatalf("backend destination=%q", params.Destination) + } + } +} + func (*releaseBackendStub) ReleaseCleanup(context.Context, backendrpc.ReleaseCleanupParams) (*backendrpc.ReleaseCleanupReport, error) { return &backendrpc.ReleaseCleanupReport{PlanID: "plan-1"}, nil } @@ -101,7 +301,7 @@ func TestReleaseHTTPForwardsTypedSelectionAndDoesNotFallback(t *testing.T) { } recorder = httptest.NewRecorder() - 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)) + server.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/v1/distribution?channel=localized&release_id=localized-1&destination=TableBundles%2FTableCatalog.bytes", nil)) if recorder.Code != http.StatusOK { t.Fatalf("distribution status=%d body=%s", recorder.Code, recorder.Body.String()) } @@ -109,8 +309,8 @@ func TestReleaseHTTPForwardsTypedSelectionAndDoesNotFallback(t *testing.T) { 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 { + backend.distributionParams[0].Offset != 0 || + backend.distributionParams[0].Limit != 0 { t.Fatalf("distribution params=%#v", backend.distributionParams) }