diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py index 85fab34b6..8643bc8aa 100755 --- a/apps/worker/worker_app/celery_app.py +++ b/apps/worker/worker_app/celery_app.py @@ -76,4 +76,10 @@ celery_app.conf.beat_schedule = { "schedule": 600.0, # 每 10 分钟(秒) "options": {"expires": 540}, }, + # 音色克隆卡死巡检:worker 重启/消息丢失后 processing 卡 10 分钟标 failed,用户可点重试 + "cleanup-stale-voice-clones": { + "task": "worker.cleanup_stale_voice_clones", + "schedule": 300.0, # 每 5 分钟 + "options": {"expires": 240}, + }, } diff --git a/apps/worker/worker_app/tasks/_startup.py b/apps/worker/worker_app/tasks/_startup.py index 8853ebafe..47b0b3a76 100644 --- a/apps/worker/worker_app/tasks/_startup.py +++ b/apps/worker/worker_app/tasks/_startup.py @@ -287,3 +287,50 @@ def _recover_stuck_ingest_jobs_on_ready(sender, **kwargs): # pragma: no cover logger.info("Worker 启动 ingest 恢复完成,共重新派单 %d 个卡死任务", recovered) except Exception as e: # noqa: BLE001 — 启动恢复失败不能阻断 worker 起服 logger.error("启动 ingest 恢复扫描失败(beat 巡检仍会兜底标 failed): %s", e, exc_info=True) + + +def recover_stale_voice_clones_on_startup(timeout_minutes: int = 10) -> int: + """Worker 启动时恢复卡死在 processing 的音色克隆任务。 + + 容器重启/进程 OOM 时 worker 中正在轮询的克隆任务会丢失, + voice_clone_profiles 永久卡在 processing 无兜底。启动时扫描 + updated_at 超过 timeout_minutes 的 processing 记录,直接标记 + 为 failed(错误信息指引用户重试)。选择标 failed 而非重新派单, + 因为 CosyVoice 侧的 voice_id 无法在无上下文下恢复轮询,重试需 + 用户确认后显式触发。 + + Args: + timeout_minutes: 判定卡死的阈值,默认 10 分钟 + + Returns: + 恢复的记录数 + """ + from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import ( + SQLAlchemyVoiceCloneProfileRepository, + ) + + try: + session = SessionLocal() + try: + repo = SQLAlchemyVoiceCloneProfileRepository(session) + count = repo.cleanup_stale_processing(timeout_minutes) + finally: + session.close() + if count > 0: + logger.warning("启动时恢复了 %d 个卡死在 processing 的音色克隆(超时 %d 分钟)", count, timeout_minutes) + else: + logger.info("无卡死 processing 音色克隆需要恢复") + return count + except Exception as e: + logger.error("启动时音色克隆恢复扫描失败(beat 巡检仍会兜底): %s", e, exc_info=True) + return 0 + + +@worker_ready.connect +def _recover_stuck_voice_clones_on_ready(sender, **kwargs): + """Worker 启动完成后恢复卡死的音色克隆任务。""" + try: + recovered = recover_stale_voice_clones_on_startup() + logger.info("Worker 启动音色克隆恢复完成,共标记 %d 个卡死任务为 failed", recovered) + except Exception as e: + logger.error("启动音色克隆恢复失败(beat 巡检仍会兜底标 failed): %s", e, exc_info=True) diff --git a/apps/worker/worker_app/tasks/cleanup.py b/apps/worker/worker_app/tasks/cleanup.py index ff29840dc..c4116c2d2 100644 --- a/apps/worker/worker_app/tasks/cleanup.py +++ b/apps/worker/worker_app/tasks/cleanup.py @@ -22,6 +22,10 @@ from packages.application.ingest_orphan_cleanup import ( INGEST_PROCESSING_TIMEOUT_MINUTES, ) +# 音色克隆 processing 超时:正常克隆轮询最多 5 分钟,10 分钟无更新视为卡死 +VOICE_CLONE_PROCESSING_TIMEOUT_MINUTES = 10 + + logger = logging.getLogger(__name__) @@ -125,3 +129,40 @@ def scheduled_cleanup_stale_ingest_jobs( purged, ) return {"stale_jobs": total_jobs, "assets_to_error": total_assets, "purged_messages": purged} + + +@shared_task(name="worker.cleanup_stale_voice_clones") +def scheduled_cleanup_stale_voice_clones( + processing_timeout_minutes: int = VOICE_CLONE_PROCESSING_TIMEOUT_MINUTES, +) -> dict: + """Celery Beat: 清理卡死在 processing 的音色克隆档案。 + + 每 5 分钟执行一次。worker 重启/Celery 消息丢失/进程 OOM 时, + 已 prefetch 的克隆任务消息丢失,voice_clone_profile 永久卡在 processing。 + 超过 processing_timeout_minutes 未更新的记录标记为 failed, + 错误信息指引用户点击重试。 + """ + from worker_app.db import SessionLocal + + from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import ( + SQLAlchemyVoiceCloneProfileRepository, + ) + + session = None + try: + session = SessionLocal() + repo = SQLAlchemyVoiceCloneProfileRepository(session) + count = repo.cleanup_stale_processing(processing_timeout_minutes) + if count > 0: + logger.warning( + "[Beat] 清理了 %d 个卡死 processing 的音色克隆(超时 %d 分钟)", + count, + processing_timeout_minutes, + ) + return {"cleaned": count} + except Exception as e: + logger.error("[Beat] 清理卡死音色克隆失败: %s", e, exc_info=True) + return {"cleaned": 0, "error": str(e)} + finally: + if session is not None: + session.close() diff --git a/apps/worker/worker_app/tasks/voice_clone.py b/apps/worker/worker_app/tasks/voice_clone.py index 69cfb1be7..4f26eac99 100755 --- a/apps/worker/worker_app/tasks/voice_clone.py +++ b/apps/worker/worker_app/tasks/voice_clone.py @@ -44,6 +44,7 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: # P2-2 修复:session 初始化为 None,避免 SessionLocal() 抛异常时 # finally 块中 session.close() 触发 UnboundLocalError session = None + logger.info(f"Voice clone task started: profile_id={profile_id}") try: session = SessionLocal() repo = SQLAlchemyVoiceCloneProfileRepository(session) diff --git a/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py b/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py index 8d4f8369f..827f09d12 100644 --- a/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py +++ b/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py @@ -136,6 +136,39 @@ class SQLAlchemyVoiceCloneProfileRepository: ) return {voice_id: profile_id for voice_id, profile_id in rows} + def cleanup_stale_processing(self, timeout_minutes: int = 10) -> int: + """清理超时卡在 processing 的克隆档案。 + + worker 重启、Celery 任务丢失或 OOM 被杀时,processing 档案会永久卡住。 + updated_at < NOW() - timeout_minutes 的 processing 记录,标记为 failed + 并附带明确错误信息,用户可在前端点击「重试」。 + + Args: + timeout_minutes: 超时分钟数,默认 10 分钟(正常克隆 < 5 分钟) + + Returns: + 清理的记录数 + """ + from datetime import UTC, datetime, timedelta + + cutoff = datetime.now(UTC) - timedelta(minutes=timeout_minutes) + models = ( + self.session.query(VoiceCloneProfileModel) + .filter( + VoiceCloneProfileModel.status == "processing", + VoiceCloneProfileModel.updated_at < cutoff, + ) + .all() + ) + count = 0 + for model in models: + model.status = "failed" + model.error_message = f"克隆任务执行超时(超过 {timeout_minutes} 分钟未更新,可能因服务重启中断),请重试" + count += 1 + if count > 0: + self.session.commit() + return count + @staticmethod def _model_to_entity(model: VoiceCloneProfileModel) -> VoiceCloneProfile: return VoiceCloneProfile( diff --git a/tests/unit/test_voice_clone_cleanup.py b/tests/unit/test_voice_clone_cleanup.py new file mode 100755 index 000000000..3ed6249f0 --- /dev/null +++ b/tests/unit/test_voice_clone_cleanup.py @@ -0,0 +1,156 @@ +"""SQLAlchemyVoiceCloneProfileRepository.cleanup_stale_processing 单元测试。 + +通过 monkeypatch sys.modules['packages.adapters.sqlalchemy_impl.models'], +注入一个具备 SQLAlchemy 列比较语义(== / < 返回可链式 .all() 的 mock)的假模型类, +不依赖真实 DB,也不会触发 SQLAlchemy 映射。 +""" + +from __future__ import annotations + +import sys +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + + +class _Col: + """模拟 SQLAlchemy Column:比较运算返回 MagicMock,可被 filter 链式调用。""" + + def __init__(self, name: str): + self._name = name + + def __eq__(self, other): # type: ignore[override] + return MagicMock(name=f"{self._name}=={other!r}") + + def __ne__(self, other): # type: ignore[override] + return MagicMock(name=f"{self._name}!={other!r}") + + def __lt__(self, other): + return MagicMock(name=f"{self._name}<{other!r}") + + def __gt__(self, other): + return MagicMock(name=f"{self._name}>{other!r}") + + def __le__(self, other): + return MagicMock(name=f"{self._name}<={other!r}") + + def __ge__(self, other): + return MagicMock(name=f"{self._name}>={other!r}") + + def __hash__(self): + return id(self) + + +class _FakeVoiceCloneProfileModel: + """假模型:类属性是 _Col;实例上可读写 status/error_message/updated_at。""" + + status = _Col("status") + updated_at = _Col("updated_at") + id = _Col("id") + error_message = _Col("error_message") + + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + +# ── 预注入 mock 模型模块,避免真实 import 拉起 DB / SQLAlchemy 映射 ── +_fake_models = SimpleNamespace(VoiceCloneProfileModel=_FakeVoiceCloneProfileModel) +sys.modules.setdefault("packages.adapters.sqlalchemy_impl.models", _fake_models) +if "packages.adapters.sqlalchemy_impl.voice_clone_profile_repository" in sys.modules: + mod = sys.modules["packages.adapters.sqlalchemy_impl.voice_clone_profile_repository"] + mod.VoiceCloneProfileModel = _FakeVoiceCloneProfileModel # type: ignore[attr-defined] + +from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import ( + SQLAlchemyVoiceCloneProfileRepository, +) + + +def _make_fake_row( + *, + status: str = "processing", + updated_at: datetime | None = None, + error_message: str = "", +) -> _FakeVoiceCloneProfileModel: + return _FakeVoiceCloneProfileModel( + status=status, + error_message=error_message, + updated_at=updated_at or datetime.now(UTC), + ) + + +def _make_repo(fake_rows: list[_FakeVoiceCloneProfileModel]): + """构造 repo + mock session。 + + 生产代码使用 .query(Model).filter(A, B).all()(一次 filter,两个表达式参数)。 + """ + session = MagicMock() + filtered = MagicMock() + filtered.all.return_value = list(fake_rows) + session.query.return_value.filter.return_value = filtered + + repo = SQLAlchemyVoiceCloneProfileRepository.__new__(SQLAlchemyVoiceCloneProfileRepository) + repo.session = session + return repo, session + + +class TestCleanupStaleProcessing: + """cleanup_stale_processing 行为测试。""" + + def test_no_stale_records_returns_zero_and_no_commit(self): + """无卡死记录时返回 0,不调用 commit。""" + repo, session = _make_repo([]) + assert repo.cleanup_stale_processing() == 0 + session.commit.assert_not_called() + + def test_stale_record_marked_failed_with_timeout_message(self): + """超时 processing 记录被标记为 failed,错误信息包含超时分钟数。""" + old = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=15)) + repo, session = _make_repo([old]) + + count = repo.cleanup_stale_processing(timeout_minutes=10) + + assert count == 1 + assert old.status == "failed" + assert "超时" in old.error_message + assert "10" in old.error_message + session.commit.assert_called_once() + + def test_error_message_reflects_custom_timeout(self): + """自定义 timeout_minutes 会反映在错误信息里。""" + old = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(hours=1)) + repo, _session = _make_repo([old]) + + repo.cleanup_stale_processing(timeout_minutes=5) + + assert old.status == "failed" + assert "5" in old.error_message + + def test_multiple_stale_records_all_cleaned_in_single_commit(self): + """多条卡死记录都被清理,返回正确计数并只 commit 一次。""" + m1 = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=20)) + m2 = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=11)) + repo, session = _make_repo([m1, m2]) + + assert repo.cleanup_stale_processing(timeout_minutes=10) == 2 + assert m1.status == "failed" + assert m2.status == "failed" + session.commit.assert_called_once() + + def test_queries_model_with_status_and_updated_at_filters(self): + """query 被调用,filter 同时传入 status=='processing' 与 updated_at