From 1994de22e0bd7ac6f208975d248aa2c83fd4aa55 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Thu, 20 Aug 2026 18:11:24 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=8E=8B=E7=BC=A9=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E5=8E=9F=E5=AD=90=E6=80=A7=20=E2=80=94=20tmp+rename=20?= =?UTF-8?q?=E6=9D=9C=E7=BB=9D=E5=8D=8A=E6=88=90=E5=93=81=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E8=A2=AB=E5=A4=8D=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因: 服务被 kill 时 dispatcher 死亡但其 ffmpeg 子进程成为孤儿 (父进程转 init)继续写压缩产物。旧代码直接写最终路径，半成品文件 size>0 且 mtime 较新，缓存复用判断(size>0 && mtime>=src)会误判 为有效 → 重试任务上传损坏视频(此前 task 300 损坏视频之谜的根源)。 多次重启还会叠加多个孤儿 ffmpeg 争抢 ARM CPU。 修复: 1. 压缩输出写 {out}.{pid}.tmp（带 PID 防多进程冲突）， 成功后 os.replace 原子 rename — 缓存目录只可能出现完整产物 2. _cleanup_compress_cache 顺带清理超过 1h 的 .tmp 残留 3. TimeoutExpired/失败路径 _safe_remove(tmp) 运维: 已清理孤儿 ffmpeg×2(task 43/44)+缓存目录，重置对应任务。 --- .../src/fam_core/dispatcher/dispatcher.py | 35 ++++++++++++++++--- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py index 67849de..81562ab 100644 --- a/fam-core/src/fam_core/dispatcher/dispatcher.py +++ b/fam-core/src/fam_core/dispatcher/dispatcher.py @@ -46,6 +46,15 @@ FFMPEG_CANDIDATES = [ ] +def _safe_remove(path): + """忽略不存在/清理失败的删除""" + try: + if os.path.isfile(path): + os.remove(path) + except OSError: + pass + + class Dispatcher: """任务下发器，30s 轮询（异步队列 + 分块断点续传）""" @@ -156,6 +165,11 @@ class Dispatcher: src_size = os.path.getsize(video_path) start = time.time() + # 写临时文件（带 PID 防多进程冲突），成功后原子 rename — + # 服务被 kill 时 ffmpeg 成为孤儿继续写 tmp，缓存目录中 + # 只会出现完整产物，杜绝半成品被复用（曾导致上传损坏视频） + tmp_path = f"{out_path}.{os.getpid()}.tmp" + cmd = [ self.ffmpeg, '-y', '-i', video_path, @@ -164,7 +178,7 @@ class Dispatcher: 'scale=trunc(iw/2)*2:trunc(ih/2)*2'), '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '28', '-an', - out_path, + tmp_path, ] try: result = subprocess.run( @@ -172,17 +186,21 @@ class Dispatcher: timeout=self.compress_timeout, ) except subprocess.TimeoutExpired: + _safe_remove(tmp_path) logger.error(f"[task_id={task_id}] 压缩超时 ({self.compress_timeout}s)，回退原始上传") return None except OSError as e: logger.error(f"[task_id={task_id}] 启动 ffmpeg 失败: {e}") return None - if result.returncode != 0 or not os.path.isfile(out_path) or os.path.getsize(out_path) == 0: + if result.returncode != 0 or not os.path.isfile(tmp_path) or os.path.getsize(tmp_path) == 0: + _safe_remove(tmp_path) stderr_tail = (result.stderr or '')[-300:] logger.error(f"[task_id={task_id}] 压缩失败 (rc={result.returncode}): {stderr_tail}") return None + os.replace(tmp_path, out_path) + dst_size = os.path.getsize(out_path) elapsed = time.time() - start logger.info(f"[task_id={task_id}] 预压缩完成: {src_size/1048576:.1f}MB → " @@ -191,7 +209,7 @@ class Dispatcher: @staticmethod def _cleanup_compress_cache(): - """清理超过 TTL 的压缩缓存目录（tmpfs 空间有限）""" + """清理超过 TTL 的压缩缓存目录 + 孤儿 .tmp 残留（tmpfs 空间有限）""" try: if not os.path.isdir(COMPRESS_DIR): return @@ -199,8 +217,15 @@ class Dispatcher: for entry in os.listdir(COMPRESS_DIR): path = os.path.join(COMPRESS_DIR, entry) try: - if os.path.isdir(path) and os.path.getmtime(path) < cutoff: - shutil.rmtree(path, ignore_errors=True) + if os.path.isdir(path): + if os.path.getmtime(path) < cutoff: + shutil.rmtree(path, ignore_errors=True) + continue + # 清理超过 1h 的 .tmp 残留（孤儿 ffmpeg 产物） + for fn in os.listdir(path): + if fn.endswith('.tmp') and \ + os.path.getmtime(os.path.join(path, fn)) < time.time() - 3600: + _safe_remove(os.path.join(path, fn)) except OSError: continue except OSError: