xiaoxia-saas/apps/worker/worker_app/tasks/cleanup.py

"""定期清理任务 — Celery Beat 调度。

包含：
- cleanup_stale_pending_tasks: 定期清理卡在 pending 超时的 generation_tasks（worker 停止消费时占位）
- cleanup_stale_running_tasks: 定期清理卡在 running 超时的 generation_tasks（容器重启/进程被杀后的孤儿）
"""

import logging

from celery import shared_task
from worker_app.tasks._startup import (
    ORPHAN_TASK_TIMEOUT_MINUTES,
    PENDING_TASK_TIMEOUT_MINUTES,
    cleanup_orphan_tasks,
    cleanup_stale_jobs,
    cleanup_stale_pending_tasks,
)

logger = logging.getLogger(__name__)


@shared_task(name="worker.cleanup_stale_pending_tasks")
def scheduled_cleanup_stale_pending(timeout_minutes: int = PENDING_TASK_TIMEOUT_MINUTES) -> dict:
    """Celery Beat 调度的定期任务：清理超时的 pending 任务。

    每 5 分钟执行一次（由 celery_app.py 的 beat_schedule 配置），
    查找所有 status='pending' 且 created_at < NOW() - timeout_minutes
    的 generation_tasks，批量更新为 failed，释放限流名额；同时 revoke 并清除
    Redis 队列中对应的 Celery 消息，杜绝作废消息重投执行（#1714）。

    Args:
        timeout_minutes: 超时时间（分钟），默认 45 分钟（pending 排队阈值放宽，
            与 running 孤儿 20 分钟区分，避免正常排队任务被误杀）

    Returns:
        {"cleaned": int}
    """
    count = cleanup_stale_pending_tasks(timeout_minutes)
    if count > 0:
        logger.info("[Beat] 清理了 %d 个超时 pending 任务（超时阈值 %d 分钟）", count, timeout_minutes)
    return {"cleaned": count}


@shared_task(name="worker.cleanup_stale_running_tasks")
def scheduled_cleanup_stale_running(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> dict:
    """Celery Beat 调度的定期任务：清理超时的 running 孤儿任务。

    每 5 分钟执行一次。worker_ready 信号只在 worker 启动时清一次，
    若 worker 没重启但任务卡死（上传挂起、进程 OOM 被内核杀掉等），
    任务会永久卡在 running 占位。此任务做持续兜底：
    查找 status='running' 且 updated_at < NOW() - timeout_minutes 的任务，
    标记为 failed（原因：容器重启/超时中断），同时清理 Job 表孤儿。

    Args:
        timeout_minutes: 超时时间（分钟），默认 20 分钟
        （worker.generate_video 硬超时 11 分钟，正常任务不可能超过 20 分钟）

    Returns:
        {"generation_tasks": int, "jobs": int}
    """
    gen_count = cleanup_orphan_tasks(timeout_minutes)
    job_count = cleanup_stale_jobs(timeout_minutes)
    total = gen_count + job_count
    if total > 0:
        logger.warning(
            "[Beat] 清理孤儿任务: running GenerationTask=%d, Job=%d（超时阈值 %d 分钟）",
            gen_count,
            job_count,
            timeout_minutes,
        )
    return {"generation_tasks": gen_count, "jobs": job_count}