Files
xiaoxia-saas/apps/worker/worker_app/celery_app.py
T
saas-backend-agent ce943faaeb
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 26s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 36s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m25s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 1m41s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m52s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 2m15s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m9s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 5m16s
AI Code Review / AI Code Review (pull_request) Successful in 6m21s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 10m15s
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 4s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 7m53s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Successful in 8s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 24s
CI/CD Pipeline / Canary Release to Production (pull_request) Failing after 267h47m29s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 267h47m32s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 267h47m29s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 267h47m32s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 267h47m32s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 267h56m18s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 267h56m18s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 267h56m18s
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Failing after 267h57m11s
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Failing after 267h57m13s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 267h57m44s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Failing after 267h57m16s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 267h57m44s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 267h57m47s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 267h57m48s
CI/CD Pipeline / Check push changed paths (pull_request) Failing after 267h57m50s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 268h30m58s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 268h31m51s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 268h32m18s
feat(generation): worker孤儿任务自动恢复+429限流结构化提示
问题:
- worker容器部署重启时,正在跑的preview/正式生成任务永久卡在pending/running
  (实测3个预览任务卡80%超10小时),孤儿任务占限流名额导致新请求429误报创建失败
- 429返回纯文本,前端无法区分「限流排队」和「创建失败」

改动:
1. worker孤儿任务巡检(apps/worker/worker_app/tasks/)
   - running孤儿清理从仅worker_ready启动时执行,扩展为beat每5分钟定时巡检
     (cleanup_stale_running_tasks),容器重启/进程OOM卡死持续兜底
   - 阈值: running 10→20分钟(任务硬超时11分钟,20分钟绝不误杀);
     pending 30→15分钟(满队列消化约10分钟,留余量)
   - pending beat巡检10→5分钟,孤儿占位释放更快
   - 失败原因结构化: error_type=WorkerInterrupted/PendingTimeout
2. 429/503限流结构化提示(app/core/task_enqueue.py)
   - build_rate_limit_detail(): code/message/queued_count/running_count/
     queue_ahead/estimated_wait_seconds/limit
   - code: USER_QUEUE_FULL(429排队等待) / SYSTEM_QUEUE_FULL(503系统繁忙)
   - 等待预估: ceil(排队数/worker并发4) × 最近20条完成任务平均耗时
   - 预览/正式生成/确认生成/重试共7处限流响应全部结构化
   - 新仓储方法鸭子类型防御,旧mock/调用方零破坏
3. 仓储新增: count_running_by_user/count_running_total/estimate_avg_duration_seconds

测试: 新增15个单测(中断running重置、stale pending重置释放名额、3孤儿批量恢复、
限流计数含预览任务、结构化detail字段、等待预估并发计算、旧仓储优雅降级);
全量14241 passed; black/isort/ruff/mypy通过
2026-09-05 11:25:57 +08:00

40 lines
1.5 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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
celery_app.conf.imports = (
"worker_app.tasks.health",
"worker_app.tasks.ingest",
"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",
"worker_app.tasks._startup",
"apps.worker.video_processing.dedup",
"worker_app.tasks.cleanup",
)
# 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},
},
}