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..df2047cb4 100644 --- a/apps/worker/worker_app/tasks/_startup.py +++ b/apps/worker/worker_app/tasks/_startup.py @@ -287,3 +287,51 @@ 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..91991635d 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,41 @@ 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..8ad36e4a8 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 datetime, timedelta, UTC + + 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(