From 243d406df045f013f8e382d6fd82724fadd747d9 Mon Sep 17 00:00:00 2001 From: hoelee Date: Mon, 12 Oct 2026 09:25:14 +0800 Subject: [PATCH] stage cleanup: purge each item's gate artefacts, sweep dead leftovers, state bug 13 Every item left its SSIM windows (291 MB each), per-offset pairs, VMAF pairs and stats logs behind; anything killed mid-item left them for ever. Measured 2026-10-12: 6 windows + 123 gate logs, work drive at 3.9 GB, both lanes idle on the guard. process() now purges its own tag/pid artefacts in the finally, and sweep_stage() clears leftovers older than STAGE_MAX_AGE_H while skipping any tag that has a fresh _out.mp4.err (i.e. is encoding right now). Regression: scripts/test_stage_cleanup.py. --- STATE.md | 17 +++++++ scripts/library_run.py | 84 +++++++++++++++++++++++++++++++++++ scripts/test_stage_cleanup.py | 73 ++++++++++++++++++++++++++++++ 3 files changed, 174 insertions(+) create mode 100644 scripts/test_stage_cleanup.py diff --git a/STATE.md b/STATE.md index 4c3c177..6abbe0c 100644 --- a/STATE.md +++ b/STATE.md @@ -260,6 +260,23 @@ Win11 默认终端是 Windows Terminal，所以看到的是"终端窗口"（类 再数可见的 `ConsoleWindowClass` + `CASCADIA_HOSTING_WINDOW_CLASS`：裸 `Popen` 会多出 1 个窗口， 带 flag 的（含全部 patched helper）保持 0。实测 2026-10-11：0→1 / 0 / 0 / 0 / 0 / 0。 +## 第 13 个 bug：闸门 raw 没人收，暂存目录越堆越满（2026-10-12） + +现象：`C:\AI\transcode-run\stage` 里积了 6 个 291 MB 的 `ssim___{ref,out}.yuv` + 123 个 `ssim_.log`。 +正常收尾只删 `_src`/`_out.mp4`，**闸门的 raw / pair / vmaf / stats 日志没人删**；被暂停、重启、看门狗杀掉的项 +留下的文件更是永远没人收。加上项目历史遗留的 45 GB 测量垃圾（`C:/AI/transcode-run/*` 与 `D:/tmp/*`）， +work 盘只剩 3.9 GB → 队列守卫拒绝所有新项，两条 lane 空转 30+ 分钟。 + +修： +1. `process()` 的 `finally` 加 `purge_item_artifacts(tag)` —— 按 tag（stage hash）与 pid 删本项的 `ssim_/pair_/vmaf_` 产物， + 另一条 lane 的文件名字里没有这些 key，绝不会误删。 +2. 新增 `sweep_stage()`，每次启动扫一遍 stage：只删老于 `STAGE_MAX_AGE_H`(6 h，`RT_STAGE_MAX_AGE_H` 可覆盖) 的残留， + 并且**整条 tag 跳过**（只要有新鲜的 `_out.mp4.err` = 那一项正在编码）。 +3. 队列侧 `rt/janitor.py`（supervisor 每 30 分钟 + **守卫失败时立刻**清一次再复检）用同一套规则把 stage 和 + `cache/runs/*.log` 收干净；`ledger*`/`*.lock`/`tsconv_state*`/`queue.sqlite*` 永不碰。 + +回归：`scripts/test_stage_cleanup.py` 6 项（另一 lane 不动 / 死残留删 / 在编 tag 保全 / 台账不动 / 字节数正确）。 + ## 红线（用户明确要求过） - **绝不删除/覆盖快照原档**（`#snapshot` 是唯一的干净档案）。 diff --git a/scripts/library_run.py b/scripts/library_run.py index b197566..ebf7134 100644 --- a/scripts/library_run.py +++ b/scripts/library_run.py @@ -160,6 +160,84 @@ def disp(p): POLICY = os.path.join(WORK, "gate_policy.json") PARK_AFTER = 2 +# A leftover in the stage dir older than this was left by a run that died (pause, reboot, the stall +# watchdog, a crash): reclaim it. Measured 2026-10-12: 6 x 291 MB SSIM windows + 123 gate logs had +# piled up, and the work drive got full enough that the guards refused every new item. +STAGE_MAX_AGE_H = float(os.environ.get("RT_STAGE_MAX_AGE_H", "6")) + + +def purge_item_artifacts(tag, workdir=None): + """Delete THIS item's gate artefacts (window raws, per-offset pairs, VMAF pairs, stats logs). + + Names carry the item tag (the stage hash) or this process's pid, so a lane running at the same + time - it shares the stage dir - can never match. Called from process()'s finally, so a clean + item leaves nothing behind even before the supervisor's janitor looks. + """ + workdir = workdir or STAGE + keys = (tag, str(os.getpid())) + removed = 0 + try: + names = os.listdir(workdir) + except OSError: + return 0 + for name in names: + if not name.startswith(("ssim_", "pair_", "vmaf_")): + continue + if not any(k and k in name for k in keys): + continue + try: + os.remove(os.path.join(workdir, name)) + removed += 1 + except OSError: + pass + return removed + + +def sweep_stage(workdir=None, max_age_h=None): + """Reclaim stage files left by runs that died. Never touches a tag that is encoding right now. + + A live item is recognised by ITS OWN progress file (`_out.mp4.err`, rewritten every 0.5 s by + ffmpeg and removed when the encode ends), so an in-flight tag is skipped wholesale. + """ + workdir = workdir or STAGE + max_age_h = STAGE_MAX_AGE_H if max_age_h is None else max_age_h + if not os.path.isdir(workdir): + return {"files": 0, "bytes": 0} + now = time.time() + cutoff = now - max_age_h * 3600 + try: + entries = os.listdir(workdir) + except OSError: + return {"files": 0, "bytes": 0} + live_tags = set() + for name in entries: + if name.endswith(".err"): + try: + if now - os.path.getmtime(os.path.join(workdir, name)) < max_age_h * 3600: + live_tags.add(name.split("_")[0]) + except OSError: + pass + files = size = 0 + for name in entries: + p = os.path.join(workdir, name) + try: + st = os.stat(p) + except OSError: + continue + if not os.path.isfile(p): + continue + if not (name.startswith(("ssim_", "pair_", "vmaf_")) or "_src." in name or "_out." in name): + continue + if st.st_mtime >= cutoff or any(t and t in name for t in live_tags): + continue + try: + os.remove(p) + files += 1 + size += st.st_size + except OSError: + pass + return {"files": files, "bytes": size} + def load_policy(): """{"accept": set(disp keys), "park": {disp key: failed rounds}} - missing file means neither.""" @@ -1104,6 +1182,9 @@ def process(src, args, dest=None, origin=None): os.remove(p) except OSError: pass + # ... and the gate's own artefacts (window raws, pairs, VMAF, stats logs): they are named + # with this item's tag or this pid, and leaving them behind is what filled the stage dir. + purge_item_artifacts(tag) def collect(root): @@ -1159,6 +1240,9 @@ def main(): DECODE_CHECK = not args.no_decode_check os.makedirs(STAGE, exist_ok=True) + swept = sweep_stage() + if swept["files"]: + log(" stage sweep: reclaimed %d dead file(s), %.0f MB" % (swept["files"], swept["bytes"] / 1e6)) if args.path: files = [args.path] if not os.path.exists(args.path): diff --git a/scripts/test_stage_cleanup.py b/scripts/test_stage_cleanup.py new file mode 100644 index 0000000..e23a5e5 --- /dev/null +++ b/scripts/test_stage_cleanup.py @@ -0,0 +1,73 @@ +"""Regression for the engine's own stage cleanup (library_run.purge_item_artifacts / sweep_stage). + +The engine shares the stage dir between lanes, so: + * purge_item_artifacts may only delete files carrying THIS item's tag or THIS process's pid, + * sweep_stage may only delete leftovers older than max_age_h, and must skip every file of a tag + that still has a fresh `_out.mp4.err` progress file (that tag is encoding right now). + + python scripts/test_stage_cleanup.py +""" +import os +import shutil +import sys +import tempfile +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import library_run as LR # noqa: E402 + +MINE, OTHER = "aa11bb22cc33", "998877665544" +root = tempfile.mkdtemp(prefix="stage_cleanup_") +OLD = 12 * 3600 + + +def touch(name, age_s=0.0, size=1024): + p = os.path.join(root, name) + with open(p, "wb") as fh: + fh.write(b"\0" * size) + if age_s: + t = time.time() - age_s + os.utime(p, (t, t)) + return p + + +# purge: only MINE / this pid +touch("ssim_%s.log" % MINE) +touch("ssim_%d_%s_ref.yuv" % (os.getpid(), MINE)) +touch("pair_a_%s.yuv" % MINE) +touch("vmaf_%s.json" % os.getpid()) +touch("ssim_%s.log" % OTHER) # another lane's item -> keep +touch("pair_a_%s.yuv" % OTHER) # another lane's item -> keep +n = LR.purge_item_artifacts(MINE, workdir=root) +left = sorted(os.listdir(root)) +purge_checks = { + "purged_mine": n == 4, + "other_lane_untouched": all(("ssim_%s.log" % OTHER) in left and ("pair_a_%s.yuv" % OTHER) in left + for _ in [0]), +} + +# sweep: old leftovers go, a live tag (fresh .err) stays whole +root2 = tempfile.mkdtemp(prefix="stage_sweep_") +root = root2 +touch("ssim_1_dead1_ref.yuv", OLD, 2048) +touch("dead99_src.mp4", OLD, 4096) +touch("ssim_2_live42_ref.yuv", OLD, 2048) # same tag as the fresh .err below -> keep +touch("live42_src.mp4", OLD, 4096) +touch("live42_out.mp4.err", 30, 256) # fresh -> tag live42 is encoding now +touch("ledger.jsonl", OLD, 8192) +st = LR.sweep_stage(workdir=root2, max_age_h=6) +left2 = sorted(os.listdir(root2)) +sweep_checks = { + "dead_leftovers_removed": "ssim_1_dead1_ref.yuv" not in left2 and "dead99_src.mp4" not in left2, + "live_tag_kept_whole": "ssim_2_live42_ref.yuv" in left2 and "live42_src.mp4" in left2, + "ledger_never_touched": "ledger.jsonl" in left2, + "freed_bytes_reported": st["bytes"] == 2048 + 4096, +} + +ok = True +for k, v in list(purge_checks.items()) + list(sweep_checks.items()): + print("%-26s %s" % (k, "PASS" if v else "FAIL")) + ok = ok and v +print("purge removed %d, sweep %s" % (n, st)) +shutil.rmtree(tempfile.gettempdir() + "/" + os.path.basename(root), ignore_errors=True) +sys.exit(0 if ok else 1)