diff --git a/alembic/versions/043_add_updated_at_to_generation_tasks.py b/alembic/versions/043_add_updated_at_to_generation_tasks.py new file mode 100755 index 000000000..24803cf53 --- /dev/null +++ b/alembic/versions/043_add_updated_at_to_generation_tasks.py @@ -0,0 +1,33 @@ +"""add updated_at to generation_tasks + +Revision ID: 043_updated_at_generation_tasks +Revises: 042_storage_key +Create Date: 2026-07-18 19:30:00.000000 + +""" + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "043_updated_at_generation_tasks" +down_revision = "042_storage_key" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "generation_tasks", + sa.Column( + "updated_at", + sa.DateTime, + nullable=False, + server_default=sa.func.now(), + ), + ) + + +def downgrade() -> None: + op.drop_column("generation_tasks", "updated_at") diff --git a/apps/api/app/api/routes/_helpers.py b/apps/api/app/api/routes/_helpers.py index d6db4fe37..a17a2a6d3 100755 --- a/apps/api/app/api/routes/_helpers.py +++ b/apps/api/app/api/routes/_helpers.py @@ -1,5 +1,6 @@ """路由层共享辅助函数 — 消除跨文件重复定义。""" +from datetime import datetime, timezone from typing import Any from fastapi import HTTPException, status @@ -122,3 +123,19 @@ def auto_select_video_assets( # 按创建时间降序(新素材在前) filtered_videos.sort(key=lambda a: a.created_at, reverse=True) return [a.id for a in filtered_videos] + + +def format_utc_datetime(dt: datetime | None) -> str: + """将数据库读出的 UTC naive datetime 格式化为带时区的 ISO 8601 字符串。 + + 数据库 DateTime 列不带时区信息,但存的是 UTC 时间。 + 直接 .isoformat() 输出无时区标识,前端会按本地时间解析,导致差 8 小时。 + 输出带 Z 后缀,前端 new Date() 自动转本地时间。 + """ + if dt is None: + return "" + if isinstance(dt, str): + return dt + if dt.tzinfo is None: + return dt.isoformat() + "Z" + return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/apps/api/app/api/routes/assets.py b/apps/api/app/api/routes/assets.py index 3c59efab7..73e029448 100755 --- a/apps/api/app/api/routes/assets.py +++ b/apps/api/app/api/routes/assets.py @@ -1,7 +1,7 @@ import logging from typing import Any, Optional -from app.api.routes._helpers import check_project_access +from app.api.routes._helpers import check_project_access, format_utc_datetime from app.auth import AuthenticatedUser, get_current_user from app.core.storage import get_storage_service from app.dependencies import ( @@ -71,7 +71,7 @@ def _to_asset_response(item, storage_service=None) -> AssetResponse: status=item.status.value, classification_status=item.classification_status.value, quality_score=item.quality_score, - created_at=item.created_at.isoformat() if item.created_at else None, + created_at=format_utc_datetime(item.created_at), uploaded_by_user_id=item.uploaded_by_user_id, tag_ids=getattr(item, "tag_ids", []), ) diff --git a/apps/api/app/api/routes/task_center.py b/apps/api/app/api/routes/task_center.py index 4804910fe..63a75a77a 100755 --- a/apps/api/app/api/routes/task_center.py +++ b/apps/api/app/api/routes/task_center.py @@ -1,6 +1,7 @@ import logging from typing import Any +from app.api.routes._helpers import format_utc_datetime from app.auth import AuthenticatedUser, get_current_user from app.core.celery_app import celery_app from app.core.task_enqueue import ( @@ -99,8 +100,8 @@ def _generation_task_to_user_response(task) -> UserTaskResponse: retryable=_status_value(task.status) == "failed", retry_count=task.retry_count or 0, source_id=task.id, - created_at=task.created_at, - updated_at=task.completed_at or task.started_at or task.created_at, + created_at=format_utc_datetime(task.created_at), + updated_at=format_utc_datetime(task.completed_at or task.started_at or task.created_at), ) @@ -119,8 +120,8 @@ def _generation_task_to_project_response(task) -> ProjectTaskResponse: retry_count=task.retry_count or 0, source_id=task.id, template_id=task.template_id, - created_at=task.created_at, - updated_at=task.completed_at or task.started_at or task.created_at, + created_at=format_utc_datetime(task.created_at), + updated_at=format_utc_datetime(task.completed_at or task.started_at or task.created_at), ) @@ -284,8 +285,8 @@ def list_project_tasks( user_message=_humanize_task_error(job.error_message), retryable=_status_value(job.status) == "failed", source_id=job.id, - created_at=job.created_at, - updated_at=job.updated_at, + created_at=format_utc_datetime(job.created_at), + updated_at=format_utc_datetime(job.updated_at), ) ) @@ -383,7 +384,7 @@ def retry_project_task( progress=0, current_step=_ingest_step(retried), source_id=retried.id, - created_at=retried.created_at, - updated_at=retried.updated_at, # type: ignore[attr-defined] + created_at=format_utc_datetime(retried.created_at), + updated_at=format_utc_datetime(retried.updated_at), # type: ignore[attr-defined] ) raise HTTPException(status_code=400, detail="Unsupported task type") diff --git a/apps/api/app/api/routes/videos.py b/apps/api/app/api/routes/videos.py old mode 100755 new mode 100644 index 8c4007e4f..9fb7e6b74 --- a/apps/api/app/api/routes/videos.py +++ b/apps/api/app/api/routes/videos.py @@ -1,5 +1,6 @@ import logging +from app.api.routes._helpers import format_utc_datetime from app.auth import AuthenticatedUser, get_current_user from app.core.celery_app import celery_app from app.core.storage import OSSStorageService, get_storage_service @@ -50,7 +51,7 @@ def _to_video_response(item, storage: OSSStorageService | None = None) -> VideoI review_status=item.review_status, generation_params=item.generation_params, download_url=download_url, - generated_at=item.generated_at.isoformat() if hasattr(item, "generated_at") and item.generated_at else "", + generated_at=format_utc_datetime(item.generated_at) if hasattr(item, "generated_at") else "", ) diff --git a/apps/api/app/schemas/task_center.py b/apps/api/app/schemas/task_center.py index c2e29f06e..70c82d449 100755 --- a/apps/api/app/schemas/task_center.py +++ b/apps/api/app/schemas/task_center.py @@ -1,5 +1,3 @@ -from datetime import datetime - from pydantic import BaseModel, Field @@ -17,8 +15,8 @@ class ProjectTaskResponse(BaseModel): retry_count: int = 0 source_id: str = "" template_id: str = "" - created_at: datetime | None = None - updated_at: datetime | None = None + created_at: str = "" + updated_at: str = "" class ListProjectTasksResponse(BaseModel): @@ -42,8 +40,8 @@ class UserTaskResponse(BaseModel): retryable: bool = False retry_count: int = 0 source_id: str = "" - created_at: datetime | None = None - updated_at: datetime | None = None + created_at: str = "" + updated_at: str = "" class ListTasksResponse(BaseModel): diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py index 7df07170b..6e51e61ff 100755 --- a/apps/worker/worker_app/celery_app.py +++ b/apps/worker/worker_app/celery_app.py @@ -17,5 +17,6 @@ celery_app.conf.imports = ( "worker_app.tasks.edit_plan_generation", "worker_app.tasks.compose_video", "worker_app.tasks.batch_download", + "worker_app.tasks._startup", "apps.worker.video_processing.dedup", ) diff --git a/apps/worker/worker_app/tasks/_startup.py b/apps/worker/worker_app/tasks/_startup.py new file mode 100755 index 000000000..8ddfd4c11 --- /dev/null +++ b/apps/worker/worker_app/tasks/_startup.py @@ -0,0 +1,55 @@ +"""Worker 启动时的初始化任务 — 孤儿任务清理等。""" + +import logging + +from worker_app.celery_app import celery_app +from worker_app.db import SessionLocal + +logger = logging.getLogger(__name__) + +ORPHAN_TASK_TIMEOUT_MINUTES = 10 + + +def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int: + """清理数据库中超时未更新的 running 任务。 + + worker 重启或崩溃后,之前处于 running 状态的任务会变成孤儿任务, + 一直卡在 running 不动。通过 updated_at 超时判断并标记为 failed。 + + Args: + timeout_minutes: 超时时间(分钟),默认 10 分钟 + + Returns: + 清理的任务数量 + """ + from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, + ) + + try: + session = SessionLocal() + repo = SQLAlchemyGenerationTaskRepository(session) + count = repo.cleanup_stale_running(timeout_minutes) + session.close() + if count > 0: + logger.warning("清理了 %d 个超时的孤儿 running 任务", count) + else: + logger.info("无孤儿 running 任务需要清理") + return count + except Exception as e: + logger.error("清理孤儿任务失败: %s", e, exc_info=True) + return 0 + + +@celery_app.on_after_configure.connect +def _setup_periodic_tasks(sender, **kwargs): + """Celery 配置完成后,注册 worker 启动钩子。""" + pass + + +@celery_app.on_worker_ready.connect +def _on_worker_ready(sender, **kwargs): + """Worker 启动完成后执行 — 清理孤儿任务。""" + logger.info("Worker 启动完成,开始清理孤儿 running 任务...") + count = cleanup_orphan_tasks() + logger.info("Worker 启动清理完成,共清理 %d 个孤儿任务", count) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 1f89202ea..1a3fc6e7e 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1409,6 +1409,16 @@ def generate_video(self, task_id: str) -> dict: logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id) + # 调度时清理一次孤儿任务(其他 worker 崩溃留下的 running 任务) + try: + from worker_app.tasks._startup import cleanup_orphan_tasks + + orphan_count = cleanup_orphan_tasks() + if orphan_count > 0: + logger.info("[task_id=%s] 调度前清理了 %d 个孤儿任务", task_id, orphan_count) + except Exception: + pass + # ── 1. 加载任务信息 ────────────────────────────────────────────────────── task_info = _load_task_info(task_id) if task_info is None: diff --git a/docs/schema-metadata-snapshot.json b/docs/schema-metadata-snapshot.json index 51ac64bce..5b1ce1dda 100644 --- a/docs/schema-metadata-snapshot.json +++ b/docs/schema-metadata-snapshot.json @@ -1628,6 +1628,14 @@ "primary_key": false, "type": "DATETIME", "unique": false + }, + { + "index": false, + "name": "updated_at", + "nullable": false, + "primary_key": false, + "type": "DATETIME", + "unique": false } ], "indexes": [ diff --git a/packages/adapters/sqlalchemy_impl/generation_task_repository.py b/packages/adapters/sqlalchemy_impl/generation_task_repository.py index 6d26a2d54..e876f538e 100755 --- a/packages/adapters/sqlalchemy_impl/generation_task_repository.py +++ b/packages/adapters/sqlalchemy_impl/generation_task_repository.py @@ -1,3 +1,5 @@ +from datetime import datetime, timezone + from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel @@ -33,6 +35,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask: batch_id=model.batch_id or "", logs=model.logs or "[]", created_at=model.created_at, + updated_at=model.updated_at, ) @@ -67,6 +70,7 @@ class SQLAlchemyGenerationTaskRepository: batch_id=task.batch_id or "", logs=task.logs, created_at=task.created_at, + updated_at=task.updated_at, ) self.session.add(model) self.session.commit() @@ -225,3 +229,37 @@ class SQLAlchemyGenerationTaskRepository: model.logs = task.logs self.session.commit() return task + + def cleanup_stale_running(self, timeout_minutes: int = 10) -> int: + """清理超时未更新的 running 任务(孤儿任务)。 + + 将 status=running 且 updated_at 超过 timeout_minutes 分钟未更新的任务 + 标记为 failed,error_message 标记为任务执行中断。 + + Returns: + 清理的任务数量 + """ + from datetime import timedelta + + cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes) + models = ( + self.session.query(GenerationTaskModel) + .filter( + GenerationTaskModel.status == GenerationTaskStatus.RUNNING.value, + GenerationTaskModel.updated_at < cutoff, + ) + .all() + ) + if not models: + return 0 + for model in models: + model.status = GenerationTaskStatus.FAILED.value + model.error_message = "任务执行中断(worker重启/超时)" + model.error_info = { + "error_type": "WorkerInterrupted", + "message": "任务在运行中中断,可能因 worker 重启或超时", + "failed_at": datetime.now(timezone.utc).isoformat(), + } + model.completed_at = datetime.now(timezone.utc) + self.session.commit() + return len(models) diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index c3d2c0446..bd90e538a 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -269,6 +269,12 @@ class GenerationTaskModel(Base): extra_meta = Column("metadata", JSON, nullable=False, default=dict) logs = Column(Text, nullable=False, default="[]", server_default="[]") created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + updated_at = Column( + DateTime, + nullable=False, + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) class GeneratedVideoModel(Base): diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py index 9b29692f3..e91134299 100755 --- a/packages/domain/generation_task.py +++ b/packages/domain/generation_task.py @@ -92,6 +92,7 @@ class GenerationTask: batch_id: str = "" logs: str = "[]" created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @classmethod def create( diff --git a/tests/unit/test_generation_task_orphan_cleanup.py b/tests/unit/test_generation_task_orphan_cleanup.py new file mode 100755 index 000000000..f9756014d --- /dev/null +++ b/tests/unit/test_generation_task_orphan_cleanup.py @@ -0,0 +1,126 @@ +"""GenerationTaskRepository - cleanup_stale_running 孤儿任务清理单元测试。""" + +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from sqlalchemy import create_engine, text +from sqlalchemy.orm import sessionmaker + +from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, +) +from packages.adapters.sqlalchemy_impl.models import Base +from packages.domain import GenerationTask, GenerationTaskStatus + + +def _repository(): + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + session = sessionmaker(bind=engine)() + return SQLAlchemyGenerationTaskRepository(session), session, engine + + +def _make_task(**kwargs) -> GenerationTask: + defaults = dict( + project_id="proj-1", + asset_library_id="lib-1", + created_by_user_id="user-1", + ) + defaults.update(kwargs) + return GenerationTask.create(**defaults) + + +def test_cleanup_stale_running_no_tasks_returns_zero(): + """没有任务时返回0。""" + repo, _, _ = _repository() + count = repo.cleanup_stale_running(timeout_minutes=10) + assert count == 0 + + +def test_cleanup_stale_running_recent_running_not_cleaned(): + """刚启动的 running 任务不清理。""" + repo, _, _ = _repository() + task = _make_task() + repo.create(task) + task.mark_processing() + repo.update(task) + + count = repo.cleanup_stale_running(timeout_minutes=10) + assert count == 0 + assert repo.get(task.id).status == GenerationTaskStatus.RUNNING + + +def test_cleanup_stale_running_old_running_marked_failed(): + """超时的 running 任务被标记为 failed。""" + repo, _, engine = _repository() + task = _make_task() + repo.create(task) + task.mark_processing() + repo.update(task) + + # 手动把 updated_at 改到 30 分钟前 + with engine.connect() as conn: + conn.execute( + text("UPDATE generation_tasks SET updated_at = :ts WHERE id = :id"), + {"ts": datetime.now(timezone.utc) - timedelta(minutes=30), "id": task.id}, + ) + conn.commit() + + count = repo.cleanup_stale_running(timeout_minutes=10) + assert count == 1 + + saved = repo.get(task.id) + assert saved.status == GenerationTaskStatus.FAILED + assert "任务执行中断" in saved.error_message + assert saved.error_info.get("error_type") == "WorkerInterrupted" + assert saved.completed_at is not None + + +def test_cleanup_stale_running_pending_not_touched(): + """pending 状态即使超时也不清理。""" + repo, _, engine = _repository() + task = _make_task() + repo.create(task) + + with engine.connect() as conn: + conn.execute( + text("UPDATE generation_tasks SET updated_at = :ts WHERE id = :id"), + {"ts": datetime.now(timezone.utc) - timedelta(hours=1), "id": task.id}, + ) + conn.commit() + + count = repo.cleanup_stale_running(timeout_minutes=10) + assert count == 0 + assert repo.get(task.id).status == GenerationTaskStatus.PENDING + + +def test_cleanup_stale_running_multiple_orphans(): + """多个超时 running 任务全部清理。""" + repo, _, engine = _repository() + + tasks = [] + for i in range(3): + t = _make_task(project_id=f"proj-{i}") + repo.create(t) + t.mark_processing() + repo.update(t) + tasks.append(t) + + # 前两个超时,第三个是新的 + with engine.connect() as conn: + for t in tasks[:2]: + conn.execute( + text("UPDATE generation_tasks SET updated_at = :ts WHERE id = :id"), + {"ts": datetime.now(timezone.utc) - timedelta(minutes=20), "id": t.id}, + ) + conn.commit() + + count = repo.cleanup_stale_running(timeout_minutes=10) + assert count == 2 + + assert repo.get(tasks[0].id).status == GenerationTaskStatus.FAILED + assert repo.get(tasks[1].id).status == GenerationTaskStatus.FAILED + assert repo.get(tasks[2].id).status == GenerationTaskStatus.RUNNING