From 509c19ce869bd7618497b4139101f952b3814903 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 14 May 2026 12:09:09 -0400 Subject: [PATCH] feat(fc2a): add maintenance tasks (recovery + cleanup) with beat schedule recover_interrupted_tasks runs every 5 minutes, finds ImportTask rows stuck in 'processing' for >30 minutes (well above any legitimate import duration), and re-queues them. cleanup_old_tasks runs daily and deletes finished tasks older than 7 days so the task table stays an operational view rather than an archive. Both thresholds match ImageRepo's precedent. The 30-min stuck threshold is documented inline so a future reader can adjust it intentionally rather than mistaking it for a 'magic number'. Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/celery_app.py | 11 +++++ backend/app/tasks/maintenance.py | 74 ++++++++++++++++++++++++++++++ tests/test_maintenance.py | 77 ++++++++++++++++++++++++++++++++ 3 files changed, 162 insertions(+) create mode 100644 backend/app/tasks/maintenance.py create mode 100644 tests/test_maintenance.py diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index bf4237e..1f10df3 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -44,6 +44,17 @@ def make_celery() -> Celery: task_acks_late=True, worker_prefetch_multiplier=1, broker_connection_retry_on_startup=True, + beat_schedule={ + "recover-interrupted-tasks": { + "task": "backend.app.tasks.maintenance.recover_interrupted_tasks", + "schedule": 300.0, # every 5 minutes + }, + "cleanup-old-tasks": { + "task": "backend.app.tasks.maintenance.cleanup_old_tasks", + "schedule": 86400.0, # daily + }, + }, + timezone="UTC", ) return app diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py new file mode 100644 index 0000000..f6e3d99 --- /dev/null +++ b/backend/app/tasks/maintenance.py @@ -0,0 +1,74 @@ +"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks.""" + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import create_engine, delete, select, update +from sqlalchemy.orm import sessionmaker + +from ..celery_app import celery +from ..config import get_config +from ..models import ImportTask + + +STUCK_THRESHOLD_MINUTES = 30 +OLD_TASK_DAYS = 7 + + +def _sync_session_factory(): + cfg = get_config() + engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True) + return sessionmaker(engine, expire_on_commit=False) + + +@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") +def recover_interrupted_tasks() -> int: + """Find ImportTask rows stuck in 'processing' for >30 min and re-queue them. + + Why 30 min: large videos can legitimately take many minutes to import; + 30 is a safe gate that catches actual crashes (which leave the row stuck + forever) without resetting slow-but-still-running jobs. + """ + SessionLocal = _sync_session_factory() + cutoff = datetime.now(timezone.utc) - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + with SessionLocal() as session: + stuck_ids = session.execute( + select(ImportTask.id) + .where(ImportTask.status == "processing") + .where(ImportTask.started_at < cutoff) + ).scalars().all() + + if not stuck_ids: + return 0 + + session.execute( + update(ImportTask) + .where(ImportTask.id.in_(stuck_ids)) + .values(status="queued", started_at=None, error="recovered from stuck state") + ) + session.commit() + + from .import_file import import_media_file + for tid in stuck_ids: + import_media_file.delay(tid) + + return len(stuck_ids) + + +@celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks") +def cleanup_old_tasks() -> int: + """Delete completed/skipped/failed ImportTask rows older than 7 days. + + Why 7 days: long enough to debug an issue an operator only notices days + later; short enough that the task table stays a useful operational view + rather than an archive. Matches IR's default. + """ + SessionLocal = _sync_session_factory() + cutoff = datetime.now(timezone.utc) - timedelta(days=OLD_TASK_DAYS) + with SessionLocal() as session: + result = session.execute( + delete(ImportTask) + .where(ImportTask.status.in_(["complete", "skipped", "failed"])) + .where(ImportTask.finished_at < cutoff) + ) + session.commit() + return result.rowcount or 0 diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py new file mode 100644 index 0000000..b38adb3 --- /dev/null +++ b/tests/test_maintenance.py @@ -0,0 +1,77 @@ +"""Unit tests for the maintenance tasks. Eager-mode Celery so the task +function runs synchronously inside the same DB transaction as the test. +""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from backend.app.celery_app import celery +from backend.app.models import ImportBatch, ImportTask + + +@pytest.fixture(autouse=True) +def eager(): + celery.conf.task_always_eager = True + yield + celery.conf.task_always_eager = False + + +def _make_batch(session) -> int: + batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick") + session.add(batch) + session.flush() + return batch.id + + +def test_recover_interrupted_only_old(db_sync): + batch_id = _make_batch(db_sync) + now = datetime.now(timezone.utc) + + fresh = ImportTask( + batch_id=batch_id, source_path="/import/a.jpg", task_type="media", + status="processing", started_at=now - timedelta(minutes=5), + ) + stale = ImportTask( + batch_id=batch_id, source_path="/import/b.jpg", task_type="media", + status="processing", started_at=now - timedelta(hours=2), + ) + db_sync.add_all([fresh, stale]) + db_sync.commit() + + from backend.app.tasks.maintenance import recover_interrupted_tasks + recovered = recover_interrupted_tasks.apply().get() + assert recovered == 1 + + db_sync.refresh(fresh) + db_sync.refresh(stale) + assert fresh.status == "processing" + assert stale.status == "queued" + assert stale.started_at is None + + +def test_cleanup_old_deletes_finished_old(db_sync): + batch_id = _make_batch(db_sync) + now = datetime.now(timezone.utc) + + old_complete = ImportTask( + batch_id=batch_id, source_path="/import/a.jpg", task_type="media", + status="complete", finished_at=now - timedelta(days=10), + ) + recent_complete = ImportTask( + batch_id=batch_id, source_path="/import/b.jpg", task_type="media", + status="complete", finished_at=now - timedelta(days=2), + ) + old_pending = ImportTask( + batch_id=batch_id, source_path="/import/c.jpg", task_type="media", + status="pending", + ) + db_sync.add_all([old_complete, recent_complete, old_pending]) + db_sync.commit() + + from backend.app.tasks.maintenance import cleanup_old_tasks + deleted = cleanup_old_tasks.apply().get() + assert deleted == 1 + + remaining = {t.source_path for t in db_sync.query(ImportTask).all()} + assert remaining == {"/import/b.jpg", "/import/c.jpg"}