diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py index 36575c3..fc50b6a 100644 --- a/backend/app/api/import_admin.py +++ b/backend/app/api/import_admin.py @@ -35,10 +35,26 @@ async def trigger_scan(): @import_admin_bp.route("/status", methods=["GET"]) async def status(): async with get_session() as session: + # Active batch = running batch that still has outstanding work. + # Plain "most recent running" picks freshly-created scans that + # enqueued zero new files and hides the older batch that's + # actually being processed. Mirrors the EXISTS predicate + # /api/system/stats already uses (api/settings.py:145-160). + # Audit 2026-06-02 — /api/import/status and /api/system/stats + # used to disagree on the active-batch predicate; the UI banner + # said "Scanning…" indefinitely while the stats card said idle. active = ( await session.execute( select(ImportBatch) - .where(ImportBatch.status == "running") + .where( + ImportBatch.status == "running", + select(ImportTask.id) + .where( + ImportTask.batch_id == ImportBatch.id, + ImportTask.status.in_(["pending", "queued", "processing"]), + ) + .exists(), + ) .order_by(ImportBatch.started_at.desc()) .limit(1) ) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index eba1a64..4051e21 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -105,6 +105,31 @@ def make_celery() -> Celery: "task": "backend.app.tasks.backup.prune_backups", "schedule": 86400.0, # daily }, + # Audit 2026-06-02 — three new per-entity recovery sweeps. + # Each runs every 5 min like the other recover_stalled_* + # sweeps; each is a no-op when nothing is stuck. + "recover-stalled-backup-runs": { + "task": "backend.app.tasks.maintenance.recover_stalled_backup_runs", + "schedule": 300.0, + }, + "recover-stalled-library-audit-runs": { + "task": "backend.app.tasks.maintenance.recover_stalled_library_audit_runs", + "schedule": 300.0, + }, + "recover-stalled-import-batches": { + "task": "backend.app.tasks.maintenance.recover_stalled_import_batches", + "schedule": 300.0, + }, + # Audit 2026-06-02 — daily retention for two entities + # whose terminal rows otherwise accumulate forever. + "prune-library-audit-runs": { + "task": "backend.app.tasks.maintenance.prune_library_audit_runs", + "schedule": 86400.0, + }, + "prune-import-batches": { + "task": "backend.app.tasks.maintenance.prune_import_batches", + "schedule": 86400.0, + }, }, timezone="UTC", ) diff --git a/backend/app/services/cleanup_service.py b/backend/app/services/cleanup_service.py index 6d104e6..28b20da 100644 --- a/backend/app/services/cleanup_service.py +++ b/backend/app/services/cleanup_service.py @@ -11,7 +11,7 @@ the one-and-done GS/IR migration tooling.) """ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any @@ -517,6 +517,9 @@ class ConfirmTokenMismatch(Exception): _VALID_RULES = ("transparency", "single_color") +_AUDIT_GUARD_THRESHOLD_MINUTES = 135 # matches LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES + + def start_audit_run( session: Session, *, rule: str, params: dict[str, Any], ) -> int: @@ -524,11 +527,21 @@ def start_audit_run( scan_library_for_rule Celery task. Returns the new audit_id. Concurrent-runs guard: raises AuditAlreadyRunning if any audit_run - has status='running'. Operator must cancel or wait.""" + has status='running' AND started recently. Audit 2026-06-02 made + the guard age-aware: a SIGKILL'd run leaves a row in 'running' + that the recovery sweep flips on its next pass (~5 min), but a + fresh start_audit_run between the SIGKILL and the sweep would + previously block forever. Past the threshold, treat the running + row as stale and let the sweep clean it up — the new run still + gets to start. + """ if rule not in _VALID_RULES: raise ValueError(f"unknown rule {rule!r}; expected one of {_VALID_RULES}") + cutoff = datetime.now(UTC) - timedelta(minutes=_AUDIT_GUARD_THRESHOLD_MINUTES) existing = session.execute( - select(LibraryAuditRun.id).where(LibraryAuditRun.status == "running") + select(LibraryAuditRun.id) + .where(LibraryAuditRun.status == "running") + .where(LibraryAuditRun.started_at >= cutoff) ).scalar_one_or_none() if existing is not None: raise AuditAlreadyRunning(existing) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index dae6f06..f58bb3d 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -177,6 +177,13 @@ class DownloadService: return {"status": "in_flight", "event_id": existing.id} if existing and existing.status == "pending": existing.status = "running" + # Reset started_at on the pending→running transition so the + # recovery sweep (DOWNLOAD_STALL_THRESHOLD_MINUTES, 30 min) + # measures from real start, not from enqueue. On heavy-queue + # days a freshly-promoted event whose original started_at + # predated the cutoff would otherwise get swept mid-flight, + # racing phase3's commit. Audit 2026-06-02. + existing.started_at = datetime.now(UTC) await self.async_session.commit() event_id = existing.id else: diff --git a/backend/app/tasks/backup.py b/backend/app/tasks/backup.py index da7c841..57d8157 100644 --- a/backend/app/tasks/backup.py +++ b/backend/app/tasks/backup.py @@ -240,8 +240,11 @@ def prune_backups() -> dict: Returns {"db_deleted": N, "images_deleted": M, "files_unlinked": K}. Tagged rows (tag IS NOT NULL) are never pruned. - Status='running' / 'restoring' rows are never pruned (recovery - sweep from FC-3i handles those via task_run). + Status='running' / 'restoring' rows are never pruned — the + recover_stalled_backup_runs sweep flips truly-stuck ones to + 'error' first. (Earlier docstring claimed the FC-3i TaskRun sweep + handled those, but TaskRun cleanup never touched BackupRun rows. + Audit 2026-06-02 added the dedicated sweep.) """ SessionLocal = _sync_session_factory() counts = {"db_deleted": 0, "images_deleted": 0, "files_unlinked": 0} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 567e97b..0ae8342 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -11,10 +11,13 @@ from sqlalchemy import Integer, and_, cast, delete, func, or_, select, update from ..celery_app import celery from ..models import ( + BackupRun, DownloadEvent, ImageRecord, + ImportBatch, ImportSettings, ImportTask, + LibraryAuditRun, Source, TaskRun, ) @@ -55,6 +58,29 @@ FFPROBE_TIMEOUT_SECONDS = 10 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days +# Audit 2026-06-02: per-entity recovery sweep thresholds. Each must be +# > the entity's longest legitimate runtime (its task's time_limit + a +# small buffer) so the sweep never flags in-flight work. +# +# Backups: images backup has time_limit=23400s (6.5h). 7h covers it +# with a 30-min buffer; db backup at 12 min hard limit fits trivially. +BACKUP_STALL_THRESHOLD_MINUTES = 7 * 60 +# Library audit: scan_library_for_rule has time_limit=7500s (2h5m). +# 2h15m gives a 10-min buffer. +LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135 +# Import batches finalize only after every child ImportTask hits a +# terminal state. The recovery sweep targets the case where every +# task is done but the batch never got its closing UPDATE +# (orchestrator crashed at the wrong instant). 2h is well past any +# realistic single-batch import. +IMPORT_BATCH_STALL_THRESHOLD_MINUTES = 120 + +# Retention windows (terminal rows older than these get deleted by +# the daily prune sweeps). 30 days = operator-flagged "useful for +# triage for a few weeks, then noise." +LIBRARY_AUDIT_KEEP_DAYS = 30 +IMPORT_BATCH_KEEP_DAYS = 30 + # Overrides for recover_stalled_task_runs (the TaskRun 'running' sweep). # Tasks/queues that legitimately run longer than the default 5-min # threshold need their own larger value, else the sweep marks in-flight @@ -69,9 +95,24 @@ TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days # files); time_limit=2100. QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { "ml": 25, + # Audit 2026-06-02 — maintenance/scan queues run tasks that + # legitimately exceed the 5-min default (verify_integrity at 70m + # hard, scan_directory at 70m hard, apply_allowlist_tags / + # recompute_centroids / backfill_phash at 35m hard). 75 min lives + # above the longest of those and the per-task overrides below + # cover the outliers (backups, library audit). + "maintenance": 75, + "scan": 75, } TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = { "backend.app.tasks.import_file.import_archive_file": 40, + # Backup images runs hours, not minutes (6.5h hard limit). The + # task-name override beats the queue's 75-min default so a + # legitimately-running backup isn't flagged. + "backend.app.tasks.backup.backup_images_task": 420, + "backend.app.tasks.backup.restore_images_task": 420, + # Library audit scans the full library — 2h hard limit. + "backend.app.tasks.library_audit.scan_library_for_rule": 130, } @@ -360,7 +401,12 @@ def prune_task_runs() -> dict: return {"ok_deleted": ok_deleted, "failures_deleted": fail_deleted} -@celery.task(name="backend.app.tasks.maintenance.backfill_phash") +@celery.task( + name="backend.app.tasks.maintenance.backfill_phash", + # Audit 2026-06-02 — keyset-paginated phash recompute over the whole + # library; legitimately runs >5 min on large libraries. + soft_time_limit=1800, time_limit=2100, +) def backfill_phash() -> int: """Recompute phash for stored images that have none (imported before FC-2d-i+ii). Keyset-paginated by id (restart-safe), NULL-only fill, @@ -438,7 +484,13 @@ def _verify_one(path: Path, expected_sha: str, mime: str, sha_fn) -> str: return "failed_verification" -@celery.task(name="backend.app.tasks.maintenance.verify_integrity") +@celery.task( + name="backend.app.tasks.maintenance.verify_integrity", + # Audit 2026-06-02 — full library sha256 + decode probe; on 100k-image + # libraries this runs an hour or more. Match the maintenance queue's + # recovery threshold (75 min) with 30s buffer below. + soft_time_limit=3600, time_limit=4200, +) def verify_integrity() -> int: """Verify every ImageRecord file: sha256 recompute + decode/probe (PIL for images; ffprobe for videos). Writes integrity_status @@ -534,6 +586,156 @@ def recover_stalled_download_events() -> int: return events_recovered +@celery.task(name="backend.app.tasks.maintenance.recover_stalled_backup_runs") +def recover_stalled_backup_runs() -> int: + """Flip BackupRun rows stuck in running/restoring past the hard limit + to error. Audit 2026-06-02. + + prune_backups (FC-3h) used to claim the FC-3i task_run sweep handled + these — but that sweep only flips TaskRun rows, not the BackupRun + artifact rows. A SIGKILL'd backup left BackupRun stuck forever + (dashboard showed phantom in-flight backups, keep_last_n offset + arithmetic skewed because zombies sat outside the ok/error window). + """ + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + cutoff = now - timedelta(minutes=BACKUP_STALL_THRESHOLD_MINUTES) + msg = ( + f"stranded by recovery sweep (no terminal status after " + f"{BACKUP_STALL_THRESHOLD_MINUTES // 60}h)" + ) + with SessionLocal() as session: + result = session.execute( + update(BackupRun) + .where(BackupRun.status.in_(["running", "restoring"])) + .where(BackupRun.started_at < cutoff) + .values(status="error", finished_at=now, error=msg) + ) + session.commit() + recovered = result.rowcount or 0 + if recovered: + log.info("recover_stalled_backup_runs: recovered %d rows", recovered) + return recovered + + +@celery.task(name="backend.app.tasks.maintenance.recover_stalled_library_audit_runs") +def recover_stalled_library_audit_runs() -> int: + """Flip LibraryAuditRun rows stuck in running past the hard limit + to error. Audit 2026-06-02. + + LibraryAuditRun.status='running' was protected by an exclusive + guard in start_audit_run — a SIGKILL'd run would block all future + audits until manual DB surgery. (The guard is now age-aware, but + this sweep is what makes that work in practice.) + """ + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + cutoff = now - timedelta(minutes=LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES) + msg = ( + f"stranded by recovery sweep (no terminal status after " + f"{LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES} min)" + ) + with SessionLocal() as session: + result = session.execute( + update(LibraryAuditRun) + .where(LibraryAuditRun.status == "running") + .where(LibraryAuditRun.started_at < cutoff) + .values(status="error", finished_at=now, error=msg) + ) + session.commit() + recovered = result.rowcount or 0 + if recovered: + log.info( + "recover_stalled_library_audit_runs: recovered %d rows", recovered, + ) + return recovered + + +@celery.task(name="backend.app.tasks.maintenance.recover_stalled_import_batches") +def recover_stalled_import_batches() -> int: + """Finalize ImportBatch rows stuck in running past the hard limit + when NO outstanding ImportTask remains. Audit 2026-06-02. + + A batch row finalizes only after every child task hits a terminal + state. The orphan case: scanner crashed between the last task's + completion and the batch's closing UPDATE. The + `/api/import/status` route then surfaces the batch as 'active' + indefinitely while `/api/system/stats` (which uses the same + EXISTS predicate we apply below) correctly returns null. + """ + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + cutoff = now - timedelta(minutes=IMPORT_BATCH_STALL_THRESHOLD_MINUTES) + with SessionLocal() as session: + # Batches still 'running' past the cutoff whose tasks are all + # terminal — there's no outstanding work, so flip the batch + # too. Mirrors the EXISTS predicate the active-batch surfaces use. + result = session.execute( + update(ImportBatch) + .where(ImportBatch.status == "running") + .where(ImportBatch.started_at < cutoff) + .where( + ~select(ImportTask.id) + .where( + ImportTask.batch_id == ImportBatch.id, + ImportTask.status.in_(["pending", "queued", "processing"]), + ) + .exists() + ) + .values(status="complete", finished_at=now) + ) + session.commit() + recovered = result.rowcount or 0 + if recovered: + log.info( + "recover_stalled_import_batches: finalized %d zombie batches", + recovered, + ) + return recovered + + +@celery.task(name="backend.app.tasks.maintenance.prune_library_audit_runs") +def prune_library_audit_runs() -> int: + """Daily retention: delete terminal LibraryAuditRun rows older than + LIBRARY_AUDIT_KEEP_DAYS. Never touches 'running'. Audit 2026-06-02. + + Audit rows carry matched_ids JSONB blobs that can hold tens of + thousands of ids; without retention these accumulate. + """ + SessionLocal = _sync_session_factory() + cutoff = datetime.now(UTC) - timedelta(days=LIBRARY_AUDIT_KEEP_DAYS) + with SessionLocal() as session: + result = session.execute( + delete(LibraryAuditRun) + .where(LibraryAuditRun.status.in_(["ready", "applied", "cancelled", "error"])) + .where(LibraryAuditRun.finished_at < cutoff) + ) + session.commit() + return result.rowcount or 0 + + +@celery.task(name="backend.app.tasks.maintenance.prune_import_batches") +def prune_import_batches() -> int: + """Daily retention: delete terminal ImportBatch rows older than + IMPORT_BATCH_KEEP_DAYS. Cascade-deletes child ImportTask rows via + the model relationship. Never touches 'running'. Audit 2026-06-02. + """ + SessionLocal = _sync_session_factory() + cutoff = datetime.now(UTC) - timedelta(days=IMPORT_BATCH_KEEP_DAYS) + with SessionLocal() as session: + # ORM-level delete here (not Core delete) so the + # ImportBatch->tasks cascade fires; Core delete would skip it. + old_batches = session.execute( + select(ImportBatch) + .where(ImportBatch.status.in_(["complete", "cancelled"])) + .where(ImportBatch.finished_at < cutoff) + ).scalars().all() + for batch in old_batches: + session.delete(batch) + session.commit() + return len(old_batches) + + @celery.task(name="backend.app.tasks.maintenance.cleanup_old_download_events") def cleanup_old_download_events() -> int: """FC-3d: delete terminal DownloadEvent rows older than the configured diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index b114eec..8867cee 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -212,7 +212,14 @@ def backfill(self) -> int: return enqueued -@celery.task(name="backend.app.tasks.ml.apply_allowlist_tags", bind=True) +@celery.task( + name="backend.app.tasks.ml.apply_allowlist_tags", + bind=True, + # Audit 2026-06-02 — the full-sweep mode (neither tag_id nor image_id) + # is O(images × allowlist) and legitimately runs >5 min on large + # libraries. Cap matches the maintenance queue's recovery threshold. + soft_time_limit=1800, time_limit=2100, +) def apply_allowlist_tags(self, tag_id: int | None = None, image_id: int | None = None) -> int: """Retroactively apply allowlisted tags. @@ -341,7 +348,13 @@ def recompute_centroid(self, tag_id: int) -> bool: return asyncio.run(_run()) -@celery.task(name="backend.app.tasks.ml.recompute_centroids", bind=True) +@celery.task( + name="backend.app.tasks.ml.recompute_centroids", + bind=True, + # Audit 2026-06-02 — drifted-centroid rebuild over potentially + # hundreds of tags. + soft_time_limit=1800, time_limit=2100, +) def recompute_centroids(self) -> int: """Daily: find drifted centroids, enqueue recompute_centroid for each.""" import asyncio diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 90398fa..7fdb1f1 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -35,7 +35,15 @@ def _iter_import_files(import_root: Path): yield entry -@celery.task(name="backend.app.tasks.scan.scan_directory", bind=True) +@celery.task( + name="backend.app.tasks.scan.scan_directory", + bind=True, + # Audit 2026-06-02 — large libraries make the scan legitimately long. + # Hard cap at 70 min so the corresponding QUEUE_STUCK_THRESHOLD_MINUTES + # ("scan") of 75 min always wins; soft limit gives the task a clean + # exit window before SIGKILL. + soft_time_limit=3600, time_limit=4200, +) def scan_directory(self, triggered_by: str = "manual", mode: str = "quick") -> int: """Walks the import root and creates ImportTasks. `mode` is 'quick'