fix: 压缩缓存原子性 — tmp+rename 杜绝半成品缓存被复用

根因: 服务被 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)+缓存目录,重置对应任务。
This commit is contained in:
ericwyuan
2026-08-20 18:11:24 +08:00
parent 47cee9db4f
commit 1994de22e0

View File

@@ -46,6 +46,15 @@ FFMPEG_CANDIDATES = [
] ]
def _safe_remove(path):
"""忽略不存在/清理失败的删除"""
try:
if os.path.isfile(path):
os.remove(path)
except OSError:
pass
class Dispatcher: class Dispatcher:
"""任务下发器30s 轮询(异步队列 + 分块断点续传)""" """任务下发器30s 轮询(异步队列 + 分块断点续传)"""
@@ -156,6 +165,11 @@ class Dispatcher:
src_size = os.path.getsize(video_path) src_size = os.path.getsize(video_path)
start = time.time() start = time.time()
# 写临时文件(带 PID 防多进程冲突),成功后原子 rename —
# 服务被 kill 时 ffmpeg 成为孤儿继续写 tmp缓存目录中
# 只会出现完整产物,杜绝半成品被复用(曾导致上传损坏视频)
tmp_path = f"{out_path}.{os.getpid()}.tmp"
cmd = [ cmd = [
self.ffmpeg, '-y', self.ffmpeg, '-y',
'-i', video_path, '-i', video_path,
@@ -164,7 +178,7 @@ class Dispatcher:
'scale=trunc(iw/2)*2:trunc(ih/2)*2'), 'scale=trunc(iw/2)*2:trunc(ih/2)*2'),
'-c:v', 'libx264', '-preset', 'veryfast', '-crf', '28', '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '28',
'-an', '-an',
out_path, tmp_path,
] ]
try: try:
result = subprocess.run( result = subprocess.run(
@@ -172,17 +186,21 @@ class Dispatcher:
timeout=self.compress_timeout, timeout=self.compress_timeout,
) )
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
_safe_remove(tmp_path)
logger.error(f"[task_id={task_id}] 压缩超时 ({self.compress_timeout}s),回退原始上传") logger.error(f"[task_id={task_id}] 压缩超时 ({self.compress_timeout}s),回退原始上传")
return None return None
except OSError as e: except OSError as e:
logger.error(f"[task_id={task_id}] 启动 ffmpeg 失败: {e}") logger.error(f"[task_id={task_id}] 启动 ffmpeg 失败: {e}")
return None 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:] stderr_tail = (result.stderr or '')[-300:]
logger.error(f"[task_id={task_id}] 压缩失败 (rc={result.returncode}): {stderr_tail}") logger.error(f"[task_id={task_id}] 压缩失败 (rc={result.returncode}): {stderr_tail}")
return None return None
os.replace(tmp_path, out_path)
dst_size = os.path.getsize(out_path) dst_size = os.path.getsize(out_path)
elapsed = time.time() - start elapsed = time.time() - start
logger.info(f"[task_id={task_id}] 预压缩完成: {src_size/1048576:.1f}MB → " logger.info(f"[task_id={task_id}] 预压缩完成: {src_size/1048576:.1f}MB → "
@@ -191,7 +209,7 @@ class Dispatcher:
@staticmethod @staticmethod
def _cleanup_compress_cache(): def _cleanup_compress_cache():
"""清理超过 TTL 的压缩缓存目录tmpfs 空间有限)""" """清理超过 TTL 的压缩缓存目录 + 孤儿 .tmp 残留tmpfs 空间有限)"""
try: try:
if not os.path.isdir(COMPRESS_DIR): if not os.path.isdir(COMPRESS_DIR):
return return
@@ -199,8 +217,15 @@ class Dispatcher:
for entry in os.listdir(COMPRESS_DIR): for entry in os.listdir(COMPRESS_DIR):
path = os.path.join(COMPRESS_DIR, entry) path = os.path.join(COMPRESS_DIR, entry)
try: try:
if os.path.isdir(path) and os.path.getmtime(path) < cutoff: if os.path.isdir(path):
shutil.rmtree(path, ignore_errors=True) 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: except OSError:
continue continue
except OSError: except OSError: