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:
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user