fix(dub): dọn phiên quá hạn xoá trúng phiên đang chạy -> job chết, trả video gốc

Lỗi người dùng gặp:
Không tổng hợp giọng được nên trả video gốc: ffmpeg ... anullsrc ...
line_0000.wav returned non-zero exit status 254
và last_error trên Space:
FileNotFoundError: '/data/tnt_work/dub_.../work/tts/line_0000.wav.mp3'

Cả engine chính, Edge đọc đỡ, lẫn bước tạo file lặng đều không ghi nổi file
-> thư mục làm việc đã BIẾN MẤT giữa job. Thủ phạm là cleanup_old: nó xét
tuổi bằng mtime của THƯ MỤC PHIÊN, mà mtime thư mục chỉ đổi khi thêm/bớt mục
con TRỰC TIẾP. Job ghi sâu bên trong (out/<video>/work/tts/...) không hề chạm
tới mtime đó, nên phiên đang xử lý vẫn bị tính là quá hạn và bị rmtree ngay
giữa chừng (sweep chạy ké mỗi lần có người mở phiên mới).

- Đếm job đang chạy theo phiên (session_in_use); cleanup_old bỏ qua phiên bận.
- cleanup_old cũng thôi đụng thư mục dùng chung (voice_previews) — không phải
phiên thì không có "hạn".
- Mọi bước ghi file TTS tự dựng lại thư mục nếu mất, để một sự cố tương tự chỉ
làm chậm chứ không giết cả job.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/api/dub.py

@@ -18,7 +18,7 @@ from fastapi import APIRouter, File, Form, HTTPException, UploadFile
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
 
-from ..core.files import new_session, rel_path, save_upload, save_upload_stream, session_dir, sub_dir
+from ..core.files import session_in_use, new_session, rel_path, save_upload, save_upload_stream, session_dir, sub_dir
 from ..core.jobs import jobs
 from ..config import settings
 from ..services.dubber import pipeline
@@ -31,6 +31,20 @@ router = APIRouter(prefix="/api/dub", tags=["dub"])
 LANGS = {"en", "vi", "fil"}
 
 
+def _submit_guarded(job, work, sid: str):
+    """Chạy job và GIỮ CHỖ phiên trong suốt lúc chạy.
+
+    Không giữ thì bộ dọn phiên quá hạn (chạy ké mỗi lần có người mở phiên mới)
+    có thể xoá đúng thư mục job đang ghi -> TTS FileNotFoundError, ffmpeg exit
+    254, job chết và chỉ trả lại video gốc.
+    """
+    def guarded(h):
+        with session_in_use(sid):
+            return work(h)
+
+    jobs.submit(job, guarded)
+
+
 @router.get("/voices")
 def voices(lang: str = "vi"):
     """Danh sách giọng + engine SẼ THẬT SỰ CHẠY cho ngôn ngữ đó.
@@ -299,7 +313,7 @@ def start(req: DubReq):
 
         return {"session": req.session, "outputs": outputs}
 
-    jobs.submit(job, work)
+    _submit_guarded(job, work, req.session)
     return {"job_id": job.id, "session": req.session, "videos": len(req.files)}
 
 
@@ -412,7 +426,7 @@ def prepare(req: DubReq):
             h.set_progress((idx + 1) / max(1, n))
         return {"session": req.session, "outputs": outputs}
 
-    jobs.submit(job, work)
+    _submit_guarded(job, work, req.session)
     return {"job_id": job.id, "session": req.session, "videos": len(req.files)}
 
 
@@ -496,7 +510,7 @@ def synthesize(req: SynthReq):
             _auto_save_dub(_username, req.session, outputs)
         return {"session": req.session, "outputs": outputs}
 
-    jobs.submit(job, work)
+    _submit_guarded(job, work, req.session)
     return {"job_id": job.id, "session": req.session, "videos": len(req.items)}
 
 
@@ -590,7 +604,7 @@ def tts_from_text(req: TtsReq):
             h.item_status(it, "error", error=str(e))
             return {"session": sid, "error": str(e)}
 
-    jobs.submit(job, work)
+    _submit_guarded(job, work, sid)
     return {"job_id": job.id, "session": sid}
 
 
@@ -640,7 +654,7 @@ def subtitle_transcribe(req: SubTranscribeReq):
             h.item_status(it, "error", error=str(e))
             return {"session": req.session, "error": str(e)}
 
-    jobs.submit(job, work)
+    _submit_guarded(job, work, req.session)
     return {"job_id": job.id, "session": req.session}
 
 
@@ -721,7 +735,7 @@ def subtitle_burn(req: SubBurnReq):
             h.item_status(it, "error", error=str(e))
             return {"session": req.session, "error": str(e)}
 
-    jobs.submit(job, work)
+    _submit_guarded(job, work, req.session)
     return {"job_id": job.id, "session": req.session}
 
 

backend/app/core/files.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 
 import os
 import shutil
+import threading
+from contextlib import contextmanager
 import time
 import uuid
 import zipfile
@@ -123,13 +125,57 @@ def delete_session(sid: str) -> bool:
     return False
 
 
+# ── Phiên ĐANG CHẠY job — cấm dọn ────────────────────────────────
+# cleanup_old xét tuổi bằng mtime của THƯ MỤC PHIÊN, mà mtime thư mục chỉ đổi
+# khi thêm/bớt mục CON TRỰC TIẾP. Job ghi file sâu bên trong (out/<video>/work/
+# tts/...) không hề chạm vào mtime đó -> một phiên đang xử lý vẫn bị coi là
+# "quá hạn" và bị xoá NGAY GIỮA JOB. Hậu quả thấy trên Space: TTS ghi file thì
+# FileNotFoundError, cả ffmpeg tạo file lặng cũng exit 254, job chết trả về
+# video gốc. Vì thế đếm job đang chạy theo phiên và bỏ qua khi dọn.
+_ACTIVE: dict[str, int] = {}
+_ACTIVE_LOCK = threading.Lock()
+
+# Thư mục trong STORAGE_DIR KHÔNG phải phiên -> đừng dọn theo tuổi phiên.
+_NOT_SESSIONS = {"voice_previews"}
+
+
+@contextmanager
+def session_in_use(sid: str):
+    """Đánh dấu phiên đang có job chạy (lồng nhau được, đếm tham chiếu)."""
+    if not sid:
+        yield
+        return
+    with _ACTIVE_LOCK:
+        _ACTIVE[sid] = _ACTIVE.get(sid, 0) + 1
+    try:
+        yield
+    finally:
+        with _ACTIVE_LOCK:
+            n = _ACTIVE.get(sid, 1) - 1
+            if n > 0:
+                _ACTIVE[sid] = n
+            else:
+                _ACTIVE.pop(sid, None)
+
+
+def session_is_busy(sid: str) -> bool:
+    with _ACTIVE_LOCK:
+        return _ACTIVE.get(sid, 0) > 0
+
+
 def cleanup_old(max_age_hours: float = 6):
-    """Dọn các phiên tạm quá hạn (mặc định 6 giờ)."""
+    """Dọn các phiên tạm quá hạn (mặc định 6 giờ).
+
+    KHÔNG đụng vào phiên đang chạy job, và không đụng thư mục dùng chung
+    (vd voice_previews) — chúng không phải phiên nên không có "hạn".
+    """
     now = time.time()
     root = settings.STORAGE_DIR
     if not os.path.isdir(root):
         return
     for name in os.listdir(root):
+        if name in _NOT_SESSIONS or session_is_busy(name):
+            continue
         p = os.path.join(root, name)
         try:
             if now - os.path.getmtime(p) > max_age_hours * 3600:

backend/app/services/dubber/tts.py

@@ -515,7 +515,20 @@ def list_voices(lang: str, allow_load: bool = False) -> list[dict]:
                     for (label, vid) in _static_voices(lang, name)]
 
 
+def _ensure_dir(path: str) -> None:
+    """Dựng lại thư mục chứa file sắp ghi nếu nó không còn.
+
+    Thư mục làm việc có thể BIẾN MẤT giữa chừng (dọn phiên quá hạn quét trúng
+    phiên đang chạy). Không dựng lại thì cả engine lẫn bước tạo file lặng đều
+    ném FileNotFoundError -> chết nguyên job, người dùng chỉ nhận lại video gốc.
+    """
+    d = os.path.dirname(os.path.abspath(path))
+    if d:
+        os.makedirs(d, exist_ok=True)
+
+
 def _run_engine(engine: TTSEngine, text: str, out_wav: str, cfg: VoiceConfig) -> str:
+    _ensure_dir(out_wav)
     engine.synth(text, out_wav, cfg)
     # Một số wrapper trả về không lỗi nhưng KHÔNG ghi file -> verify để bước sau
     # (fit_to_slot) không FileNotFound làm vỡ pipeline.
@@ -574,6 +587,7 @@ def is_silent_wav(path: str) -> bool:
 
 
 def _silence(out_wav: str, dur: float):
+    _ensure_dir(out_wav)
     subprocess.run(
         [FFMPEG, "-y", "-hide_banner", "-loglevel", "error",
          "-f", "lavfi", "-i", "anullsrc=r=24000:cl=mono",

backend/tests/test_files.py

@@ -90,3 +90,45 @@ def test_cleanup_old_removes_aged():
     fresh = files.new_session("download")  # phiên mới phải còn nguyên
     files.cleanup_old(max_age_hours=6)
     assert os.path.isdir(files.session_dir(fresh))
+
+def test_cleanup_spares_busy_session_and_shared_dirs():
+    """Dọn phiên quá hạn KHÔNG được đụng phiên đang chạy job.
+
+    Lỗi thật trên Space: mtime của thư mục phiên chỉ đổi khi thêm/bớt mục con
+    TRỰC TIẾP, mà job ghi file sâu bên trong (out/<video>/work/tts/...) nên
+    phiên đang xử lý vẫn bị coi là quá hạn -> xoá giữa chừng, TTS ném
+    FileNotFoundError và ffmpeg tạo file lặng cũng exit 254.
+    """
+    import os
+    import time
+    from app.core import files as F
+
+    root = settings.STORAGE_DIR
+    busy = os.path.join(root, "dub_busy_000000")
+    idle = os.path.join(root, "dub_idle_000000")
+    shared = os.path.join(root, "voice_previews")
+    for d in (busy, idle, shared):
+        os.makedirs(os.path.join(d, "out", "v", "work", "tts"), exist_ok=True)
+        old = time.time() - 48 * 3600
+        os.utime(d, (old, old))                 # cả ba đều "quá hạn"
+
+    with F.session_in_use("dub_busy_000000"):
+        F.cleanup_old(6)
+        assert os.path.isdir(busy), "phiên đang chạy job bị xoá mất"
+    assert os.path.isdir(shared), "thư mục dùng chung không phải phiên, đừng xoá"
+    assert not os.path.isdir(idle), "phiên rảnh quá hạn thì vẫn phải dọn"
+
+    F.cleanup_old(6)                            # hết job -> mới được dọn
+    assert not os.path.isdir(busy)
+
+
+def test_session_in_use_nested():
+    """Hai job cùng phiên -> chỉ hết bận khi job cuối xong."""
+    from app.core import files as F
+
+    with F.session_in_use("s1"):
+        with F.session_in_use("s1"):
+            assert F.session_is_busy("s1")
+        assert F.session_is_busy("s1")           # job ngoài vẫn chạy
+    assert not F.session_is_busy("s1")
+