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,
)

from packages.application.ingest_orphan_cleanup import (
    ASSET_ORPHAN_TIMEOUT_MINUTES,
    INGEST_PENDING_TIMEOUT_MINUTES,
    INGEST_PROCESSING_TIMEOUT_MINUTES,
)

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}


@shared_task(name="worker.cleanup_stale_ingest_jobs")
def scheduled_cleanup_stale_ingest_jobs(
    processing_timeout_minutes: int = INGEST_PROCESSING_TIMEOUT_MINUTES,
    pending_timeout_minutes: int = INGEST_PENDING_TIMEOUT_MINUTES,
    orphan_asset_timeout_minutes: int = ASSET_ORPHAN_TIMEOUT_MINUTES,
) -> dict:
    """Celery Beat 调度：清理上传/转码链路（IngestJob + Asset）孤儿记录。

    每 10 分钟执行一次。worker 容器重启/进程 OOM 时，已 prefetch 的 transcode
    celery 消息会丢失（队列里也不存在），ingest_job 永久卡 pending/processing、
    asset 永久卡 processing/uploading，没有兜底永远不会恢复（#1714）。

    - ingest_job processing > processing_timeout_minutes / pending > pending_timeout_minutes
      → 标 failed；关联 asset 占位（processing/uploading）联动标 error
    - 无 ingest_job 关联、created_at > orphan_asset_timeout_minutes 的占位 asset
      → 标 error
    - 作废 celery 消息 revoke + 物理清除（防重投，执行前守卫是第二道防线）
    """
    from worker_app.db import SessionLocal

    from packages.application.ingest_orphan_cleanup import (
        cleanup_orphan_processing_assets,
        cleanup_stale_ingest_jobs,
        revoke_stale_ingest_messages,
    )

    session = SessionLocal()
    try:
        job_items, asset_ids = cleanup_stale_ingest_jobs(
            session,
            processing_timeout_minutes=processing_timeout_minutes,
            pending_timeout_minutes=pending_timeout_minutes,
        )
        orphan_asset_ids = cleanup_orphan_processing_assets(session, timeout_minutes=orphan_asset_timeout_minutes)
    finally:
        session.close()

    purged = revoke_stale_ingest_messages(job_items) if job_items else 0
    total_jobs = len(job_items)
    total_assets = len(set(asset_ids) | set(orphan_asset_ids))
    if total_jobs or total_assets:
        logger.warning(
            "[Beat] 清理 ingest 链路孤儿: stale_jobs=%d, assets→error=%d, 队列清除消息=%d",
            total_jobs,
            total_assets,
            purged,
        )
    return {"stale_jobs": total_jobs, "assets_to_error": total_assets, "purged_messages": purged}