From 3a612919f74f8a131cf355e96921dc79c72c990c Mon Sep 17 00:00:00 2001 From: cat <1716967236@qq.com> Date: Fri, 11 Sep 2026 16:12:32 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=9D=E6=8C=81=E4=BA=A7=E7=89=A9?= =?UTF-8?q?=E6=94=B6=E5=B0=BE=E5=B9=82=E7=AD=89=E5=B9=B6=E4=BF=9D=E6=8A=A4?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E6=88=90=E5=93=81=E5=AE=8C=E6=95=B4=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 13 ++- docs/代码审查问题跟踪.md | 14 ++- src/wov_app/batch.py | 17 ++-- src/wov_app/scheduler.py | 57 ++++++++----- src/wov_app/storage.py | 26 ++++++ tests/test_batch.py | 36 +++++--- tests/test_finalization.py | 169 +++++++++++++++++++++++++++++++++++++ tests/test_scheduler.py | 4 +- 8 files changed, 293 insertions(+), 43 deletions(-) create mode 100644 src/wov_app/storage.py create mode 100644 tests/test_finalization.py diff --git a/AGENTS.md b/AGENTS.md index 7c30fcd..a174934 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -135,8 +135,13 @@ whisper 节点按以下顺序解析模型路径，默认避免从远端下载： 等假名，真实短对话（そこ/やばい/ねえ/やだ/えへへ）天然不命中。仅 decode_full 生效，demo 等 VAD 链路不受影响。 -最终产物按 `上传文件名.标识.时间戳` 重命名（如 `test01.zh-CN.20260815123000.srt`）， -标识优先取节点的 `target_language` 参数，否则用产物别名。 +最终产物按 `上传文件名.标识.时间戳` 命名（如 `test01.zh-CN.20260815123000.srt`）， +标识优先取节点的 `target_language` 参数，否则用产物别名。**审查 R03 修复**： +保留节点原始文件及 URI，把成品副本存入 `runs//finals/output-<编码别名>/`； +时间戳固定取 run 创建时间，重复收尾覆盖相同路径，多个别名分目录避免冲突。 +复制先写同目录临时文件，再原子替换目标，失败不登记残缺文件；`final_outputs` +声明的引用或文件缺失时任务失败，不能标完成。旧版本原文件已改名但最终别名记录 +仍指向有效文件时允许复用；原文件与成品都丢失时明确报错。 ### 任务参数覆盖（前端框选） @@ -356,6 +361,10 @@ http://127.0.0.1:8000/docs API 文档 存为 `<视频名>.CN.srt`、双目字幕存为 `<视频名>.CN_dual_eye.ass`（稳定无时间戳， `_sidecar_product_name` 映射，其余扩展名产物保留原文件名；同名目标直接覆盖）。 文件名含视频主名，下次批量扫描会命中"已有字幕"规则直接跳过该视频。 +- **成品放置校验（审查 R03）**：先预检全部 `final_outputs` 对应的记录与文件， + 缺任一项即失败，不开始覆盖视频旁成品；全部齐备后逐文件原子替换。复制失败时 + 视频记 FAILED，保留 run、工作空间和已放置的完整成品，供修复后幂等重试。 + 原子替换保证单文件完整，不代表多个成品的跨文件事务或断电持久性。 - **过程文件清理（2026-09 起）**：视频收尾完成后删除该视频的整个工作空间与 run 记录（音频/分块/帧图/节点产物不留残），防止媒体库把切片数据当视频入库。 工作空间位于**应用私有目录** `data/storage/batch///` diff --git a/docs/代码审查问题跟踪.md b/docs/代码审查问题跟踪.md index 6413ce2..38584bf 100644 --- a/docs/代码审查问题跟踪.md +++ b/docs/代码审查问题跟踪.md @@ -10,7 +10,7 @@ | --- | --- | --- | --- | --- | | R01 | P0 | 普通删除接口可能删除批量源视频目录；清理逻辑从输入路径推导删除范围 | 已修复 | 普通接口拒绝单独删除批量 run；手动与自动清理只删除该上传任务的私有目录；视频、旁挂字幕及其他任务文件不受影响 | | R02 | P1 | 自适应线程池限流后可能扩容；退出标记位于积压队列尾部，缩容不及时 | 已修复 | 限流不增加并发；降低目标后不再超额提交；真实积压队列验证 | -| R03 | P1 | 最后节点执行时暂停再继续会覆盖有效下载 URI；批量缺产物仍清理并完成 | 待处理 | 收尾幂等，恢复后全部必需产物可下载；缺产物不清理工作空间 | +| R03 | P1 | 最后节点执行时暂停再继续会覆盖有效下载 URI；批量缺产物仍清理并完成 | 已修复 | 收尾幂等，恢复后全部必需产物可下载；缺产物不清理工作空间 | | R04 | P1 | 环形 DAG 可通过保存校验，执行失败后仍在 QUEUED 堵塞队列 | 待处理 | 保存/发布拒绝无效 DAG；历史无效任务进入 FAILED，不阻塞后续任务 | | R05 | P1 | 翻译按固定四行解析 SRT；补齐行数不能保证文本与时间轴对应 | 待处理 | 合法多行 cue 正确解析；按稳定 ID 回填译文并校验缺失、重复项 | | R06 | P1 | OCR 跨空白帧合并相同字幕；临时请求失败被永久存为无文字 | 待处理 | 空白帧结束当前字幕段；无文字和可重试失败分开存档 | @@ -59,6 +59,18 @@ - 补充重试额度用例后：`uv run pytest tests/test_adaptive_pool.py -q`，22 passed，0.27 秒。连续限流后的第二轮 map 保持 1 并发，进度序号重新从 1 开始。`git diff --check` 通过。 - 状态：修复及验证完成，纳入 `fix/review-improvements` 分支；未部署。尚未对真实服务配额下的吞吐做性能结论；已发出请求的终止依赖节点自身超时。 +## R03 修复记录 + +- 根因：最终产物原地改名后，节点输出 URI 未同步，恢复时旧路径覆盖有效最终记录；批量放置对缺失产物直接跳过，随后仍清理并标记完成。 +- 实现：[scheduler.py](../src/wov_app/scheduler.py) 保留节点原文件，把命名成品复制至 run 的 finals 别名目录；时间戳固定取 run 创建时间，重复收尾路径稳定。缺少最终引用或文件时进入 FAILED。 +- 兼容：旧版本已经改名但最终别名记录仍有效时复用该路径，不再覆盖为失效节点 URI；原始与最终文件均丢失时明确报错。 +- 文件操作：[storage.py](../src/wov_app/storage.py) 提供同目录临时文件复制和原子替换，失败清理临时文件、保留源及原目标。[batch.py](../src/wov_app/batch.py) 先预检全部必需成品，复制成功后才清理 run 和工作空间；中途失败保留断点供重试。 +- TDD 红：新增/强化的 8 个场景失败，覆盖暂停恢复链接、旧改名记录、多别名同源、缺少输出引用/文件、批量缺产物清理、节点原始 URI 保留及复制故障。 +- 相关回归：`uv run pytest tests/test_finalization.py tests/test_scheduler.py tests/test_batch.py tests/test_apps_api.py tests/test_db.py -q --tb=short`，97 passed，3.22 秒。 +- 补充批量复制故障及安全回归：`uv run pytest tests/test_finalization.py tests/test_maintenance.py tests/test_run_deletion_safety.py -q --tb=short`，22 passed，5.83 秒。真实视频/字幕验证写满故障后旧成品和工作空间完整，故障解除可成功放置并清理；API 下载验证原始文件与两个最终别名均可读取。仅有既存 Starlette/httpx 弃用警告。 +- 状态：修复及验证完成，纳入 `fix/review-improvements` 分支；未部署。原子性为单文件级，多文件中途失败允许部分完整成品已经更新，重试会重新放置；不承诺断电持久性。 +- 剩余边界：批量固定工作流版本仍由 R08 跟踪；此修复不会自动重建已经丢失的历史成品。 + ## 审查基线 - 修复前全套测试：`uv run pytest`，369 passed、6 skipped，76.75 秒。 diff --git a/src/wov_app/batch.py b/src/wov_app/batch.py index 6c01803..fed52e1 100644 --- a/src/wov_app/batch.py +++ b/src/wov_app/batch.py @@ -48,6 +48,7 @@ from wov_app.config import BATCH_INTERVAL_SECONDS, STORAGE_DIR from wov_app.db import Database from wov_app.logging import get_logger from wov_app.scheduler import WorkflowScheduler +from wov_app.storage import atomic_copy from wov_sdk.models import WorkflowDefinition # 批量引擎运行日志：任务进度、视频逐个处理与暂停/续跑等状态变化。 @@ -531,18 +532,22 @@ class BatchWorker: "已有字幕"规则跳过该视频。同名目标直接覆盖：可能是上一次运行/旧工作流 留下的旧内容，应以本次产物为准。 """ - placed: list[str] = [] - video.parent.mkdir(parents=True, exist_ok=True) + # 先校验全部必需输出，缺文件时不覆盖任何视频旁成品，更不能继续清理。 + products: list[tuple[Path, Path]] = [] for alias in definition.final_outputs: artifact = self.db.get_artifact(run_id, alias) if artifact is None: - continue + raise ValueError(f"missing final artifact: {alias}") source = Path(artifact["uri"]) if not source.is_file(): - continue + raise ValueError(f"missing final artifact file: {alias} ({source})") target = video.parent / _sidecar_product_name(video, source) - # 目标名稳定 → 直接覆盖写入，避免旧同名产物被"大小一致复用"误保留。 - shutil.copy2(source, target) + products.append((source, target)) + placed: list[str] = [] + for source, target in products: + # 稳定目标名 + 原子替换：复制失败保留旧成品，异常交调用方记录 FAILED。 + # 多文件中途失败仍保留完整工作空间，下次可以幂等地重新放置。 + atomic_copy(source, target) placed.append(target.name) return placed diff --git a/src/wov_app/scheduler.py b/src/wov_app/scheduler.py index 6c7fe2d..b910f87 100755 --- a/src/wov_app/scheduler.py +++ b/src/wov_app/scheduler.py @@ -18,6 +18,7 @@ from wov_app import registry from wov_app.config import SCHEDULER_INTERVAL_SECONDS from wov_app.db import Database from wov_app.logging import get_logger +from wov_app.storage import atomic_copy # 调度器运行日志：节点进度、暂停/续跑等状态变化。 logger = get_logger("scheduler") @@ -237,19 +238,21 @@ class WorkflowScheduler: # 处理 final_outputs，为用户端提供简洁的下载别名。 for alias, ref in definition.final_outputs.items(): resolved = self._resolve_ref(ref, run.get("input_uri"), outputs_by_node) - if resolved is not None: - # 最终产物按 上传文件名.标识.时间戳 重命名，区分语言与版本。 - resolved = self._final_artifact_uri(resolved, run, definition, alias, ref) - self.db.create_artifact( - { - "run_id": run_id, - "node_id": ref.partition(".")[0], - "name": alias, - "uri": resolved, - "mime_type": self._mime_type(resolved), - "size": self._file_size(resolved), - } - ) + if resolved is None: + raise ValueError(f"missing final output: {alias} ({ref})") + # 成品按 上传文件名.标识.时间戳 命名，保留节点原始文件与 URI。 + # 同一个 run 重复收尾使用稳定路径，不再因暂停/重启反复改名。 + resolved = self._final_artifact_uri(resolved, run, definition, alias, ref) + self.db.create_artifact( + { + "run_id": run_id, + "node_id": ref.partition(".")[0], + "name": alias, + "uri": resolved, + "mime_type": self._mime_type(resolved), + "size": self._file_size(resolved), + } + ) # 全部节点成功后标记完成；期间被暂停则保持 PAUSED，等待续跑补做收尾。 if self.db.get_run(run_id)["status"] == "PAUSED": @@ -285,25 +288,35 @@ class WorkflowScheduler: alias: str, ref: str, ) -> str: - """把最终产物重命名为 上传文件名.标识.时间戳 并返回新 URI。 + """生成命名成品副本并返回稳定 URI，节点原始产物始终保留。 - 标识优先取产出节点的 target_language 参数（如 zh-CN），否则回退为 - 产物别名；时间戳取当前时刻，用于区分同一上传文件的多次运行版本。 - 重命名在原地进行（同目录），不复制文件。 + 标识优先取产出节点的 target_language，否则回退别名；时间戳固定为 + run 创建时间。每个别名单独目录，防止相同语言/扩展名的输出互相覆盖。 + 复制使用原子替换；收尾重复执行覆盖相同目标，不生成新的时间戳副本。 """ source = Path(resolved) - # 续跑等场景下源文件可能已被上次收尾重命名过：不再重命名，原样返回。 if not source.is_file(): - return resolved + # 兼容旧版本：原文件已改名，但最终别名仍记录有效路径时直接复用。 + existing = self.db.get_artifact(run["id"], alias) + if existing is not None and Path(existing["uri"]).is_file(): + return existing["uri"] + raise ValueError(f"missing final output file: {alias} ({resolved})") # 基础名来自上传文件名；无上传文件时退回通用名称 subtitle。 base = Path(run["input_uri"]).stem if run.get("input_uri") else "subtitle" # 通过最终输出引用定位产出节点，取其语言参数作为标识。 node_id = ref.partition(".")[0] node = next((item for item in definition.nodes if item.id == node_id), None) tag = (node.params.get("target_language") if node else None) or alias - timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") - new_path = source.with_name(f"{base}.{tag}.{timestamp}{source.suffix}") - source.rename(new_path) + # 编码目录别名避免路径分隔符；保留常规 cn_srt/ass 名称便于排查。 + from urllib.parse import quote + + timestamp = datetime.fromisoformat(run["created_at"]).strftime("%Y%m%d%H%M%S") + final_dir = self.storage_dir / "runs" / run["id"] / "finals" / ("output-" + quote(alias, safe="")) + filename = f"{base}.{tag}.{timestamp}{source.suffix}" + if Path(filename).name != filename or "\\" in filename: + raise ValueError(f"invalid final output filename: {alias}") + new_path = final_dir / filename + atomic_copy(source, new_path) return str(new_path) @staticmethod diff --git a/src/wov_app/storage.py b/src/wov_app/storage.py new file mode 100644 index 0000000..dc59f58 --- /dev/null +++ b/src/wov_app/storage.py @@ -0,0 +1,26 @@ +"""产物文件操作：以同目录临时文件复制并原子替换，避免暴露写到一半的成品。""" + +from __future__ import annotations + +import shutil +import tempfile +from pathlib import Path + + +def atomic_copy(source: Path, target: Path) -> None: + """保留源文件，完整复制后替换目标；失败清理临时文件并保留原目标。 + + 临时文件与目标位于同一目录，确保 replace 不跨文件系统；关闭句柄后再 + copy2/replace，兼容 Windows。原子性针对单个文件，不承诺多文件事务或 + 断电持久性；调用方仅在全部必需文件复制成功后登记完成并清理工作空间。 + """ + target.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + dir=target.parent, prefix=f".{target.name}.", suffix=".tmp", delete=False + ) as handle: + temporary = Path(handle.name) + try: + shutil.copy2(source, temporary) + temporary.replace(target) + finally: + temporary.unlink(missing_ok=True) diff --git a/tests/test_batch.py b/tests/test_batch.py index 2eb8d39..762a3a3 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -682,13 +682,16 @@ def test_batch_worker_pending_leftover_never_marks_completed(tmp_path, monkeypat monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _CrashScheduler) # 放开假调度器的崩溃：第二次调用时不再删 run，让真实调度器跑通。 class _RecoveringScheduler: - """第二轮：不再崩溃，把 run 置 COMPLETED（由引擎收尾放产物）。""" + """第二轮：真实执行工作流并产生成品，由引擎收尾放置。""" def __init__(self, db: Database, work_dir: Path) -> None: self.db = db def execute_run(self, run_id: str) -> None: - self.db.update_run(run_id, status="COMPLETED", updated_at=_now_iso()) + from wov_app.scheduler import WorkflowScheduler + + video = self.db.list_batch_videos(job["id"])[0] + WorkflowScheduler(self.db, Path(video["work_dir"])).execute_run(run_id) monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _RecoveringScheduler) worker._process_job(job) @@ -701,7 +704,7 @@ def test_batch_worker_already_completed_runs_finalized(tmp_path) -> None: """run 已完成但视频未标记（收尾前中断）：直接放置产物、清理并标记完成。 覆盖 _place_products 各分支：产物齐全（复制）、产物记录存在但文件丢失 - （跳过）、无产物记录（跳过）——完成后均清理工作空间与 run 记录。 + （报错）、无产物记录（报错）——只有成品齐全才清理工作空间与 run 记录。 """ db = _db(tmp_path) _seed_echo_workflow(db) @@ -714,6 +717,7 @@ def test_batch_worker_already_completed_runs_finalized(tmp_path) -> None: """创建 COMPLETED run；可选产物文件与产物记录。""" _create_run(db, run_id, folder, status="COMPLETED", name=name) work_dir = Path(videos[name]["work_dir"]) + work_dir.mkdir(parents=True) if with_artifact: path = work_dir / "runs" / run_id / "steps" / "step" / f"{name}.result.txt" db.create_artifact( @@ -739,20 +743,24 @@ def test_batch_worker_already_completed_runs_finalized(tmp_path) -> None: BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"])) videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])} - assert all(videos[name]["status"] == "COMPLETED" for name in ("a", "b", "c")) + assert videos["a"]["status"] == "COMPLETED" + assert all(videos[name]["status"] == "FAILED" for name in ("b", "c")) # a 的产物复制到视频旁；b/c 无产物可复制。 product = _find_product(folder, "a", "result") assert product.read_text(encoding="utf-8") == "产物 a" assert not list(folder.glob("b.result.*")) and not list(folder.glob("c.result.*")) - # 三个视频的工作空间与 run 记录都被清理。 - for name in ("a", "b", "c"): - assert not Path(videos[name]["work_dir"]).exists() - assert db.get_run(f"run_done_{name}") is None - assert db.list_runs() == [] + # a 清理完成；b/c 必须保留工作空间、关联 run 与错误供修复后重试。 + assert not Path(videos["a"]["work_dir"]).exists() + assert db.get_run("run_done_a") is None + for name in ("b", "c"): + assert Path(videos[name]["work_dir"]).exists() + assert db.get_run(f"run_done_{name}") is not None + assert videos[name]["run_id"] == f"run_done_{name}" + assert "result" in videos[name]["error"] def test_batch_worker_place_products_branches(tmp_path) -> None: - """_place_products：无产物记录/文件丢失跳过；.srt/.ass 按库内约定命名覆盖； + """_place_products：无产物记录/文件丢失报错；.srt/.ass 按库内约定命名覆盖； 其他扩展名保留原文件名复制。""" db = _db(tmp_path) _seed_echo_workflow(db) @@ -805,6 +813,14 @@ def test_batch_worker_place_products_branches(tmp_path) -> None: {"run_id": run_id, "node_id": "step", "name": "alias_txt", "uri": str(src_txt), "mime_type": "text/plain", "size": src_txt.stat().st_size} ) + # 所有必需成品预检通过前不开始覆盖，缺失原因中包含具体别名。 + with pytest.raises(ValueError, match="alias_none"): + worker._place_products(run_id, video, WorkflowDefinition.from_dict(definition)) + del definition["final_outputs"]["alias_none"] + with pytest.raises(ValueError, match="alias_lost"): + worker._place_products(run_id, video, WorkflowDefinition.from_dict(definition)) + assert target_srt.read_text(encoding="utf-8") == "旧字幕内容" + del definition["final_outputs"]["alias_lost"] placed = worker._place_products(run_id, video, WorkflowDefinition.from_dict(definition)) assert placed == ["a.CN.srt", "a.CN_dual_eye.ass", "a.other.20260902120000.txt"] # srt/ass 被改名为标准名且覆盖旧文件；txt 保留原名。 diff --git a/tests/test_finalization.py b/tests/test_finalization.py new file mode 100644 index 0000000..22d2f97 --- /dev/null +++ b/tests/test_finalization.py @@ -0,0 +1,169 @@ +"""产物收尾回归：真实 SQLite、字幕文件和 API 下载验证暂停及故障恢复。""" + +import shutil +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from wov_app.db import Database +from wov_app.main import app +from wov_app.scheduler import WorkflowScheduler +from wov_sdk.models import WorkflowDefinition + + +@pytest.fixture +def subtitle(tmp_path): + """复用真实字幕资产，不在测试中生成模型输出或占位媒体。""" + source = Path(__file__).resolve().parent.parent / "testdata/ocr_srt_run_ac7f480a3ccb.srt" + if not source.is_file(): + pytest.skip("缺少真实字幕资产") + target = tmp_path / "original.srt" + shutil.copy2(source, target) + return target + + +def _seed(db, source, finals): + """登记已完成节点的真实产物，模拟节点执行结束后的断点。""" + definition = {"name": "finalization", "version": 1, + "nodes": [{"id": "step", "node_type": "echo"}], + "final_outputs": finals} + db.upsert_workflow({"id": "finalization", "name": "收尾回归"}) + db.create_workflow_version("finalization", 1, definition) + db.create_run({"id": "run_finalization", "workflow_id": "finalization", + "workflow_version": 1, "status": "QUEUED", "input_uri": str(source), + "created_at": "2026-01-01T00:00:00+00:00", "updated_at": "2026-01-01T00:00:00+00:00"}) + db.create_artifact({"run_id": "run_finalization", "node_id": "step", + "name": "step.file_uri", "uri": str(source)}) + return WorkflowDefinition.from_dict(definition) + + +def test_pause_during_final_registration_keeps_downloads(subtitle, tmp_path, monkeypatch): + """在第一个最终别名登记后暂停，再继续，两别名均可下载且 URI 不漂移。""" + db = Database(tmp_path / "test.db") + _seed(db, subtitle, {"result": "step.file_uri", "copy": "step.file_uri"}) + original = subtitle.read_bytes() + create = db.create_artifact + + def pause_once(artifact): + create(artifact) + if artifact["name"] == "result": + db.pause_run("run_finalization", "2026-01-01T00:00:00+00:00") + + scheduler = WorkflowScheduler(db, tmp_path / "storage") + with monkeypatch.context() as patch: + patch.setattr(db, "create_artifact", pause_once) + scheduler.execute_run("run_finalization") + first_uri = db.get_artifact("run_finalization", "result")["uri"] + assert db.get_run("run_finalization")["status"] == "PAUSED" + db.resume_run("run_finalization", "2026-01-01T00:00:00+00:00") + scheduler.execute_run("run_finalization") + assert db.get_run("run_finalization")["status"] == "COMPLETED" + assert db.get_artifact("run_finalization", "result")["uri"] == first_uri + with TestClient(app) as client: + monkeypatch.setattr(app.state, "db", db) + for alias in ("result", "copy", "step.file_uri"): + response = client.get(f"/api/runs/run_finalization/artifacts/{alias}") + assert response.status_code == 200 + assert response.content == original + + +def test_legacy_renamed_final_survives_resume(subtitle, tmp_path): + """兼容旧版已改名但最终记录有效的断点，恢复不能覆盖成不存在的旧路径。""" + db = Database(tmp_path / "test.db") + _seed(db, subtitle, {"result": "step.file_uri"}) + renamed = subtitle.with_name("legacy.srt") + subtitle.rename(renamed) + db.create_artifact({"run_id": "run_finalization", "node_id": "step", + "name": "result", "uri": str(renamed)}) + WorkflowScheduler(db, tmp_path / "storage").execute_run("run_finalization") + assert db.get_run("run_finalization")["status"] == "COMPLETED" + assert Path(db.get_artifact("run_finalization", "result")["uri"]).read_bytes() == renamed.read_bytes() + + +def test_failed_final_copy_preserves_source_and_can_retry(subtitle, tmp_path, monkeypatch): + """复制中途写入失败不登记残缺成品，源字幕可用，恢复后可成功收尾。""" + db = Database(tmp_path / "test.db") + _seed(db, subtitle, {"result": "step.file_uri"}) + expected = subtitle.read_bytes() + scheduler = WorkflowScheduler(db, tmp_path / "storage") + + def fail_copy(source, destination, **kwargs): + # 文件 I/O 边界模拟磁盘写满：真实写出截断内容后报错。 + Path(destination).write_bytes(Path(source).read_bytes()[:64]) + raise OSError("disk full") + + with monkeypatch.context() as patch: + patch.setattr(shutil, "copy2", fail_copy) + scheduler.execute_run("run_finalization") + assert db.get_run("run_finalization")["status"] == "FAILED" + assert db.get_artifact("run_finalization", "result") is None + assert subtitle.read_bytes() == expected + assert not list((tmp_path / "storage").rglob("*.tmp")) + db.update_run("run_finalization", status="QUEUED", updated_at="2026-01-01T00:00:00+00:00") + scheduler.execute_run("run_finalization") + assert db.get_run("run_finalization")["status"] == "COMPLETED" + assert Path(db.get_artifact("run_finalization", "result")["uri"]).read_bytes() == expected + + +def test_batch_copy_failure_preserves_old_product_and_workspace(subtitle, tmp_path, monkeypatch): + """批量成品写到一半失败：旧字幕不变、run/工作空间保留，重试可收尾。""" + from wov_app.batch import BatchWorker, create_job + + video_asset = Path(__file__).resolve().parent.parent / "testdata/subtitle_10s.mp4" + if not video_asset.is_file(): + pytest.skip("缺少真实视频资产") + library = tmp_path / "library" + library.mkdir() + video = library / "movie.mp4" + shutil.copy2(video_asset, video) + db = Database(tmp_path / "test.db") + _seed(db, subtitle, {"result": "step.file_uri"}) + db.upsert_workflow({"id": "finalization", "name": "收尾回归", "published": 1, "latest_version": 1}) + job_id = create_job(db, str(library), "finalization") + item = db.list_batch_videos(job_id)[0] + work_dir = Path(item["work_dir"]) + work_dir.mkdir(parents=True) + source = work_dir / "result.srt" + shutil.copy2(subtitle, source) + now = "2026-01-01T00:00:00+00:00" + db.update_batch_video(item["id"], run_id="run_finalization", updated_at=now) + db.update_run("run_finalization", status="COMPLETED", updated_at=now) + db.create_artifact({"run_id": "run_finalization", "node_id": "step", + "name": "result", "uri": str(source)}) + target = library / "movie.CN.srt" + shutil.copy2(subtitle, target) + original = target.read_bytes() + worker = BatchWorker(db) + + def fail_copy(source, destination, **kwargs): + Path(destination).write_bytes(Path(source).read_bytes()[:64]) + raise OSError("disk full") + + with monkeypatch.context() as patch: + patch.setattr(shutil, "copy2", fail_copy) + worker._process_job(db.get_batch_job(job_id)) + assert db.get_batch_video(item["id"])["status"] == "FAILED" + assert target.read_bytes() == original + assert db.get_run("run_finalization") is not None + assert source.read_bytes() == original + assert not list(library.glob("*.tmp")) + worker._process_job(db.get_batch_job(job_id)) + assert db.get_batch_video(item["id"])["status"] == "COMPLETED" + assert db.get_run("run_finalization") is None + assert not work_dir.exists() + assert target.read_bytes() == original + assert video.read_bytes() == video_asset.read_bytes() + + +@pytest.mark.parametrize("missing_ref", [False, True]) +def test_missing_final_fails_run(subtitle, tmp_path, missing_ref): + """必需最终输出无引用或文件丢失都应 FAILED，不能成功登记失效链接。""" + db = Database(tmp_path / "test.db") + _seed(db, subtitle, {"result": "step.missing" if missing_ref else "step.file_uri"}) + if not missing_ref: + subtitle.unlink() + WorkflowScheduler(db, tmp_path / "storage").execute_run("run_finalization") + run = db.get_run("run_finalization") + assert run["status"] == "FAILED" + assert "result" in run["error"] diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 9092c23..fdaaf38 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -429,9 +429,9 @@ def test_final_artifact_renamed_with_language_tag(tmp_path) -> None: assert filename.startswith("movie01.zh-CN.") assert filename.endswith(".txt") assert Path(final["uri"]).is_file() - # 原始未重命名文件不应残留。 + # 节点原始 URI 必须保持有效；成品副本与原始文本一致，保证断点可恢复。 step_artifacts = [item for item in artifacts if item["name"] == "step.file_uri"] - assert not Path(step_artifacts[0]["uri"]).exists() + assert Path(step_artifacts[0]["uri"]).read_bytes() == Path(final["uri"]).read_bytes() def test_final_artifact_renamed_fallback_base_and_tag(tmp_path) -> None: