diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 61a46dd..b6ce64c 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -81,6 +81,14 @@ def make_celery() -> Celery: "task": "backend.app.tasks.maintenance.cleanup_old_download_events", "schedule": 86400.0, # daily }, + "recover-stalled-task-runs": { + "task": "backend.app.tasks.maintenance.recover_stalled_task_runs", + "schedule": 300.0, # every 5 min, matches recover-interrupted-tasks + }, + "prune-task-runs": { + "task": "backend.app.tasks.maintenance.prune_task_runs", + "schedule": 86400.0, # daily + }, }, timezone="UTC", ) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 2a50e86..1a2e028 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -9,7 +9,7 @@ from PIL import Image from sqlalchemy import delete, select, update from ..celery_app import celery -from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask +from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun from ..utils.phash import compute_phash from ._sync_engine import sync_session_factory as _sync_session_factory @@ -20,6 +20,8 @@ OLD_TASK_DAYS = 7 PHASH_PAGE = 500 VERIFY_PAGE = 200 FFPROBE_TIMEOUT_SECONDS = 10 +TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h +TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") @@ -80,6 +82,71 @@ def cleanup_old_tasks() -> int: return result.rowcount or 0 +@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs") +def recover_stalled_task_runs() -> int: + """Flip task_run rows stuck in 'running' for >STUCK_THRESHOLD_MINUTES + to 'error'. FC-3i. + + A row gets stuck when the worker dies without emitting + task_postrun / task_failure (e.g. OOM, container restart between + signals, signal handler raised+logged). Shares the 5-min threshold + with recover_interrupted_tasks for consistency. + """ + SessionLocal = _sync_session_factory() + cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + with SessionLocal() as session: + result = session.execute( + update(TaskRun) + .where(TaskRun.status == "running") + .where(TaskRun.started_at < cutoff) + .values( + status="error", + error_type="RecoverySweep", + error_message=( + f"no completion signal received within " + f"{STUCK_THRESHOLD_MINUTES} min" + ), + finished_at=datetime.now(UTC), + ) + ) + session.commit() + return result.rowcount or 0 + + +@celery.task(name="backend.app.tasks.maintenance.prune_task_runs") +def prune_task_runs() -> dict: + """Daily retention for task_run rows. FC-3i. + + - 'ok' rows: deleted after TASK_RUN_KEEP_OK_SECONDS (24h default). + Success is high-volume, not interesting after a day. + - 'error' / 'timeout' rows: deleted after TASK_RUN_KEEP_FAILURE_SECONDS + (7 days default). Failures are operationally interesting longer. + - 'running' rows: NEVER deleted by this task. The recovery sweep + (recover_stalled_task_runs) is the mechanism that flips them to + terminal state; prune doesn't touch in-flight state. + - 'retry' rows: treated as failures (>7d). + + Returns dict of how many rows were deleted in each bucket. + """ + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + ok_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_OK_SECONDS) + fail_cutoff = now - timedelta(seconds=TASK_RUN_KEEP_FAILURE_SECONDS) + with SessionLocal() as session: + ok_deleted = session.execute( + delete(TaskRun) + .where(TaskRun.status == "ok") + .where(TaskRun.finished_at < ok_cutoff) + ).rowcount or 0 + fail_deleted = session.execute( + delete(TaskRun) + .where(TaskRun.status.in_(["error", "timeout", "retry"])) + .where(TaskRun.finished_at < fail_cutoff) + ).rowcount or 0 + session.commit() + return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted} + + @celery.task(name="backend.app.tasks.maintenance.backfill_phash") def backfill_phash() -> int: """Recompute phash for stored images that have none (imported before