CI Bot
a33532113b
根因: - 声音克隆走 API提交CosyVoice→Celery worker轮询5分钟→mark ready/failed - 当worker容器重启、进程OOM或Celeryprefetch消息丢失时,已进入processing状态的voice_clone_profile没有兜底恢复机制 - 对比ingest链路有cleanup_stale_ingest_jobs beat任务+worker_ready启动恢复,voice_clone链路缺失 - 结果: 用户看到"克隆处理中，请稍候..."永久转圈,刷新也不变 修复: 1. SQLAlchemyVoiceCloneProfileRepository新增cleanup_stale_processing方法: 扫描updated_at超过10分钟的processing记录(正常克隆<5分钟),标记failed并提示用户重试 2. _startup.py新增recover_stale_voice_clones_on_startup+worker_ready信号: worker启动时一次性扫描恢复,用户重启/部署后立即解锁卡死任务 3. cleanup.py新增scheduled_cleanup_stale_voice_clones beat任务: 每5分钟巡检兜底,防止运行期OOM/消息丢失导致的新卡死 4. celery_app.py beat_schedule注册新任务 5. voice_clone.py任务入口加received日志,便于日志排查
86 lines
4.3 KiB
Python
Executable File
86 lines
4.3 KiB
Python
Executable File
from celery import Celery
|
||
from worker_app.core.config import get_settings
|
||
|
||
settings = get_settings()
|
||
celery_app = Celery(settings.worker_name)
|
||
celery_app.conf.broker_url = settings.broker_url
|
||
celery_app.conf.result_backend = settings.result_backend
|
||
celery_app.conf.broker_connection_retry_on_startup = True
|
||
|
||
# #1714 队列隔离：generation（高优，独占 worker）/ transcode（素材转码）/ celery（默认）
|
||
from packages.shared.celery_queues import ( # noqa: E402
|
||
GENERATION_WORKER_PREFETCH_MULTIPLIER,
|
||
apply_queue_settings,
|
||
)
|
||
|
||
apply_queue_settings(celery_app)
|
||
# 长渲染任务预取 1，避免任务被预取占住导致调度不均
|
||
celery_app.conf.worker_prefetch_multiplier = GENERATION_WORKER_PREFETCH_MULTIPLIER
|
||
celery_app.conf.task_acks_late = True # worker 崩溃时未完成任务重回队列，由执行前守卫丢弃作废消息
|
||
# worker 进程被 OOM/容器硬杀时拒绝 ack，消息留在队列由其他 worker 接手
|
||
celery_app.conf.task_reject_on_worker_lost = True
|
||
# Redis broker 消息可见性超时（#1714）：acks_late 下，消息被预取后 visibility_timeout
|
||
# 内未 ack 才会重投。长任务（ingest HEVC 转码 20-30 分钟、生成硬超时 11 分钟）
|
||
# 必须远大于最长执行时间，否则正常任务会在执行中被误重投；4 小时覆盖最长转码 + 余量。
|
||
celery_app.conf.broker_transport_options = {"visibility_timeout": 4 * 60 * 60}
|
||
|
||
celery_app.conf.imports = (
|
||
"worker_app.tasks.health",
|
||
"worker_app.tasks.ingest",
|
||
"worker_app.tasks.atom_clips",
|
||
# #1970 片段级 AI 标签：必须显式 import 注册，否则 worker 报
|
||
# "Received unregistered task of type 'worker.tag_atom_clip'"
|
||
"worker_app.tasks.atom_clip_tagging",
|
||
"worker_app.tasks.asset_quality_scoring_task",
|
||
"worker_app.tasks.backfill_atom_clip_tags",
|
||
"worker_app.tasks.classification",
|
||
"worker_app.tasks.generation",
|
||
"worker_app.tasks.voice_extraction",
|
||
"worker_app.tasks.voice_clone",
|
||
"worker_app.tasks.tts_synthesis",
|
||
"worker_app.tasks.batch_download",
|
||
"worker_app.tasks.duplication_check",
|
||
# #1798 AI 数字人渲染：必须在 Worker 实例上注册同名任务，否则消息无人消费（渲染卡 0%）
|
||
"worker_app.tasks.ai_avatar_render",
|
||
"worker_app.tasks._startup",
|
||
"apps.worker.video_processing.dedup",
|
||
"worker_app.tasks.cleanup",
|
||
# 注意：必须用 app.* 路径，不能用 apps.api.app.* 路径！
|
||
# PYTHONPATH=/app/apps/api 下，app.tasks.lipsync_tts 可直接导入且不触发 apps/api/__init__.py
|
||
# （apps/api/__init__.py 会 from .main import app，级联加载整个 FastAPI 栈，Worker 中不需要且会导致注册失败）
|
||
"app.tasks.lipsync_tts",
|
||
# #1998 GPU MuseTalk 异步推理：wait_for_result→签名 URL→回写 lipsync_jobs
|
||
# 必须在 Worker 侧注册，否则 apply_async 消息无人消费，job 永远卡在 processing
|
||
"app.tasks.lipsync_gpu",
|
||
)
|
||
|
||
# Celery Beat 定时任务调度
|
||
# 注：worker 单实例内嵌 beat（entrypoint-worker.sh -B），定时任务不会重复执行
|
||
celery_app.conf.beat_schedule = {
|
||
# pending 任务超时清理：worker 停止消费后，卡 pending 的任务 15 分钟内释放限流名额
|
||
"cleanup-stale-pending-tasks": {
|
||
"task": "worker.cleanup_stale_pending_tasks",
|
||
"schedule": 300.0, # 每 5 分钟（秒）
|
||
"options": {"expires": 240}, # 4 分钟过期，避免堆积
|
||
},
|
||
# running 孤儿任务巡检：容器重启/进程被杀后卡 running 的任务，20 分钟无更新则判失败
|
||
"cleanup-stale-running-tasks": {
|
||
"task": "worker.cleanup_stale_running_tasks",
|
||
"schedule": 300.0, # 每 5 分钟（秒）
|
||
"options": {"expires": 240},
|
||
},
|
||
# 上传/转码链路孤儿巡检：worker 重启丢 prefetch 消息后，卡 pending/processing
|
||
# 的 ingest_job + asset 占位超时标终态（#1714）。转码任务较长，10 分钟一轮
|
||
"cleanup-stale-ingest-jobs": {
|
||
"task": "worker.cleanup_stale_ingest_jobs",
|
||
"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},
|
||
},
|
||
}
|