"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks.""" import logging import os import subprocess from datetime import UTC, datetime, timedelta from pathlib import Path from PIL import Image from sqlalchemy import Integer, and_, cast, delete, func, or_, select, update from ..celery_app import celery from ..models import ( BackupRun, DownloadEvent, HeadAutoApplyRun, HeadTrainingRun, ImageRecord, ImportBatch, ImportSettings, ImportTask, LibraryAuditRun, Source, TagEvalRun, TaskRun, ) from ..utils.phash import compute_phash from ._sync_engine import get_sync_engine from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) # High-churn tables whose dead-tuple bloat matters: the TABLESAMPLE showcase # reads physical blocks (bloat slows it directly), and the periodic # prune/backfill/recovery tasks generate dead tuples faster than autovacuum # always keeps up with. VACUUM reclaims them; ANALYZE refreshes planner stats. # Allowlist ONLY — names are interpolated into VACUUM, so they must never come # from request input. VACUUM_TABLES = ( "image_record", "image_provenance", "post_attachment", "download_event", "task_run", "import_task", "import_batch", ) STUCK_THRESHOLD_MINUTES = 5 # Archive ImportTasks run the per-member pipeline inline for every # member (import_archive_file: soft=30min/hard=35min). The ImportTask # 'processing' recovery sweep must give them a longer threshold or it # re-queues a legitimately-running archive mid-import (double-process). # 40 min = 5-min buffer past the archive task's hard kill. # Operator-flagged 2026-05-28 (target 1645019, a big archive). ARCHIVE_STUCK_THRESHOLD_MINUTES = 40 # Poison-pill cap. After being recovered (re-queued from a stuck # 'processing' state) MAX_RECOVERY_ATTEMPTS-1 times, the next sweep # marks the row 'failed' instead of looping. 3 = two recoveries then # give up. A row reaches this only if it leaves NO terminal flip each # run — i.e. it hard-crashes the worker (OOM/segfault/SIGKILL), the # signature of a corrupt or oversized input. Caught exceptions already # flip to terminal 'failed' and never enter this loop. MAX_RECOVERY_ATTEMPTS = 3 ORPHAN_PENDING_THRESHOLD_MINUTES = 30 # DownloadEvent (pending|running) recovery threshold. download_source has # time_limit=1500s (25 min, DOWNLOAD_HARD_TIME_LIMIT); 30 min is 5 min past # that, so a legitimately-running task is hard-killed before the sweep ever # touches it — the sweep only catches events whose worker died without # finalizing. Operator-confirmed 2026-05-29 after 43 sources stranded at # "last check never" by the in-flight guard; budget bumped 2026-06-03 with # the soft/hard limit raise (Anduo #39912). DOWNLOAD_STALL_THRESHOLD_MINUTES = 30 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 # 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. BACKUP_STALL_THRESHOLD_MINUTES = 7 * 60 # DB backup/restore is seconds-to-minutes (35-min hard limit). It must NOT share # the images' 7h window — a DB backup wedged on NFS would otherwise sit "running" # for 7 hours holding the concurrency-1 maintenance_long lane (operator-flagged # 2026-06-07). 40 min gives a small buffer over the hard limit. BACKUP_DB_STALL_THRESHOLD_MINUTES = 40 # Library audit: scan_library_for_rule has time_limit=7500s (2h5m). # 2h15m gives a 10-min buffer. LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135 # tag-eval (#1130) has a 30-min soft limit; flag a run with no progress past 40. TAG_EVAL_STALL_THRESHOLD_MINUTES = 40 TAG_EVAL_KEEP_RUNS = 20 # head training (#114) has a 60-min soft limit; flag no-progress past 75. HEAD_TRAINING_STALL_THRESHOLD_MINUTES = 75 HEAD_TRAINING_KEEP_RUNS = 20 # head auto-apply (#114) shares the 60-min soft limit; flag past 75. HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES = 75 HEAD_AUTO_APPLY_KEEP_RUNS = 20 # 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 # work 'error' before it finishes. Each value MUST be ≥ the relevant # task.time_limit + a small buffer. task_name overrides take precedence # over queue overrides. # # ml queue: tag_and_embed video branch (≈20 GPU ops); time_limit=1200. # import_archive_file: shares the 'import' queue with the fast # single-file import_media_file, so it needs a task-name override # (the import queue itself stays at the 5-min default for single # files); time_limit=2100. QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { "ml": 25, # download_source legitimately walks 5-25 min (Patreon/gallery-dl # deep creators); its hard time_limit is DOWNLOAD_HARD_TIME_LIMIT # (1500s = 25m). The 5-min default flagged healthy in-flight walks as # phantom 'RecoverySweep' failures (System Activity showed errors the # Subscriptions view correctly didn't — the download finished ok and # reset the source). 30 clears the 25-min limit with buffer and lines # up with DOWNLOAD_STALL_THRESHOLD_MINUTES (30) so a genuine hard kill # is swept by the task-run AND event sweeps together. Audit 2026-06-10. "download": 30, # 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, # External file-host fetches (mega/gdrive/film packs) can run to the task's # 60-min hard limit (time_limit=3600) — the fetcher's own read/total budgets # (external_fetch) cap a single fetch below that, but this stays the outer # backstop. Without an override these healthy in-flight fetches were # phantom-flagged 'RecoverySweep' before their own timeout/error could # surface (operator-flagged 2026-06-17, target 414 swept at 6.6min). A # task-name override beats the queue threshold whatever queue the row records # (it recorded 'default' before the celery_signals fix → download). 65 = 60+5. "backend.app.tasks.external.fetch_external_link": 65, } @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") def recover_interrupted_tasks() -> int: """Recover stuck ImportTask rows. Two distinct stuck states: 1. 'processing' too long — worker crash mid-import. Re-queue via enqueue_import (routing media vs archive) and let the import retry. Threshold is task-type-aware: media files are sub-second and capped at the 5-min soft limit, so STUCK_THRESHOLD_MINUTES (5) means a confirmed crash; archives run the per-member pipeline inline (import_archive_file, 35-min hard limit) so they get ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid re-queueing a still-running archive. (Media was tightened from 30 min to 5 2026-05-24 after a 2224-row zombie pile; archive split out 2026-05-28.) 2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory creates rows with status='pending' (commit), then in a second pass transitions to 'queued' and calls .delay() (commit). If the scanner crashes between those two commits, rows are orphaned in 'pending' (never enqueued) with no recovery path — invisible to the 'processing' sweep above. Flagged 2026-05-25 by operator hitting a 5490-row orphan pile. Flip these to 'failed' (not re-enqueue) so the operator drains them via /api/import/retry-failed at their own pace; bulk-re-enqueueing 5000+ rows would thundering-herd the import worker. Returns total rows touched (recovered + marked failed). """ SessionLocal = _sync_session_factory() now = datetime.now(UTC) media_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES) archive_cutoff = now - timedelta(minutes=ARCHIVE_STUCK_THRESHOLD_MINUTES) orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES) with SessionLocal() as session: # Both sweeps used to be SELECT ids → UPDATE WHERE id IN (...) which # blew past psycopg's 65535-parameter ceiling once a sweep covered # tens of thousands of rows (operator hit it 2026-05-26 after the # /import deep scan piled up orphans). Folding the SELECT into the # UPDATE eliminates the IN-list entirely. RETURNING gives us back # exactly the (id, task_type) pairs that flipped so the requeue # can route media vs archive correctly. # # Media + archive get separate cutoffs: a single media file is # sub-second so 5 min means crash; an archive runs the per-member # pipeline inline and can legitimately take up to its 35-min hard # limit, so it gets ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid # re-queueing a still-running archive. stuck_predicate = and_( ImportTask.status == "processing", or_( and_(ImportTask.task_type != "archive", ImportTask.started_at < media_cutoff), and_(ImportTask.task_type == "archive", ImportTask.started_at < archive_cutoff), ), ) # POISON-PILL CIRCUIT BREAKER (Layer 1, 2026-05-28). A row that # leaves no terminal flip (hard worker crash: OOM/segfault/SIGKILL # on a corrupt or oversized input) gets re-queued by this sweep — # and would loop forever, re-crashing the worker each pass, # without a cap. Once a row has already been recovered # MAX_RECOVERY_ATTEMPTS-1 times, stop re-queueing it and mark it # 'failed' with a diagnostic so the operator can find + replace # the offending file. This UPDATE runs FIRST so the rows it # claims drop out of 'processing' before the re-queue pass. poison_result = session.execute( update(ImportTask) .where(stuck_predicate) .where(ImportTask.recovery_count >= MAX_RECOVERY_ATTEMPTS - 1) .values( status="failed", finished_at=now, error=( f"crashed or stalled the worker {MAX_RECOVERY_ATTEMPTS} " f"times without completing — likely a corrupt or " f"oversized input. Not re-queued. Inspect/replace the " f"file, then retry via /api/import/retry-failed." ), ) .returning(ImportTask.id) ) poison_ids = [r[0] for r in poison_result.all()] # Re-queue the remaining stuck rows (under the cap) and bump # their recovery_count. RETURNING (id, task_type) so the requeue # routes media vs archive correctly. stuck_result = session.execute( update(ImportTask) .where(stuck_predicate) .where(ImportTask.recovery_count < MAX_RECOVERY_ATTEMPTS - 1) .values( status="queued", started_at=None, recovery_count=ImportTask.recovery_count + 1, error="recovered from stuck state", ) .returning(ImportTask.id, ImportTask.task_type) ) stuck = stuck_result.all() orphan_result = session.execute( update(ImportTask) .where(ImportTask.status.in_(["pending", "queued"])) .where(ImportTask.created_at < orphan_cutoff) .values( status="failed", # Without finished_at, cleanup_old_tasks (`WHERE # finished_at < cutoff`) never reaps these rows — # orphan-swept rows would become permanent table # tenants. Audit 2026-06-02. finished_at=now, error=( "orphan pending/queued swept by recover_interrupted_tasks " "(scanner likely crashed mid-enqueue); retry via " "/api/import/retry-failed" ), ) ) orphan_count = orphan_result.rowcount or 0 session.commit() if stuck: from .import_file import enqueue_import for tid, task_type in stuck: enqueue_import(tid, task_type) # Layer-2 auto re-download (env-gated, default OFF). For each # poison-pill row that resolves to a pollable Source, delete the # bad file and trigger ONE source re-check to fetch a fresh # copy. Bounded by ImportTask.refetched so source-side # corruption can't loop. The 'failed' row stays as history; the # re-downloaded file re-imports as a fresh task on the next scan. if poison_ids and os.environ.get("FC_AUTO_REFETCH_CORRUPT", "0") == "1": from ..models import ImportSettings from ..services.refetch_service import attempt_refetch import_root = Path(session.execute( select(ImportSettings.import_scan_path) .where(ImportSettings.id == 1) ).scalar_one()) for pid in poison_ids: ptask = session.get(ImportTask, pid) if ptask is None: continue try: attempt_refetch(session, ptask, import_root) except Exception as exc: # noqa: BLE001 — best-effort log.warning("auto-refetch failed for task %s: %s", pid, exc) return len(stuck) + len(poison_ids) + orphan_count @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(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 @celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs") def recover_stalled_task_runs() -> int: """Flip task_run rows stuck in 'running' past their queue-specific threshold 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). The default 5-min threshold fits short-lived queues (import/thumbnail/download); queues that legitimately run longer tasks (ml-video, deep scans) get their own larger threshold via QUEUE_STUCK_THRESHOLD_MINUTES so the sweep doesn't preempt them. Runs once per distinct threshold value: each pass updates rows whose queue maps to that threshold. """ SessionLocal = _sync_session_factory() now = datetime.now(UTC) override_tasks = set(TASK_STUCK_THRESHOLD_MINUTES.keys()) override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys()) total = 0 def _flag(minutes, *extra_where): cutoff = now - timedelta(minutes=minutes) stmt = ( 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 {minutes} min" ), finished_at=now, # Matches celery_signals.finalize's # int((now - started_at).total_seconds() * 1000) # — sweep-closed rows now carry duration like # normally-finalized rows. Audit 2026-06-02. duration_ms=cast( func.extract("epoch", now - TaskRun.started_at) * 1000, Integer, ), ) ) for w in extra_where: stmt = stmt.where(w) return session.execute(stmt).rowcount or 0 with SessionLocal() as session: # Precedence: task_name override → queue override → default. # Each pass excludes rows claimed by a higher-precedence pass so # every row is touched at most once. # 1. Per-task-name overrides (e.g. import_archive_file, which # shares the 'import' queue with fast single-file imports). for task_name, minutes in TASK_STUCK_THRESHOLD_MINUTES.items(): total += _flag(minutes, TaskRun.task_name == task_name) # 2. Per-queue overrides, excluding the override task-names. for queue, minutes in QUEUE_STUCK_THRESHOLD_MINUTES.items(): wheres = [TaskRun.queue == queue] if override_tasks: wheres.append(TaskRun.task_name.notin_(override_tasks)) total += _flag(minutes, *wheres) # 3. Default — everything not claimed above. default_wheres = [] if override_queues: default_wheres.append(TaskRun.queue.notin_(override_queues)) if override_tasks: default_wheres.append(TaskRun.task_name.notin_(override_tasks)) total += _flag(STUCK_THRESHOLD_MINUTES, *default_wheres) session.commit() return total @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", # 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, idempotent. Videos legitimately keep phash NULL. A missing/unreadable file is logged and left NULL — never fails the task.""" SessionLocal = _sync_session_factory() updated = 0 last_id = 0 with SessionLocal() as session: while True: rows = session.execute( select(ImageRecord) .where(ImageRecord.id > last_id) .where(ImageRecord.phash.is_(None)) .where(ImageRecord.mime.like("image/%")) .order_by(ImageRecord.id.asc()) .limit(PHASH_PAGE) ).scalars().all() if not rows: break for rec in rows: try: with Image.open(rec.path) as im: ph = compute_phash(im) except Exception as exc: log.warning( "backfill_phash: unreadable %s: %s", rec.path, exc ) ph = None if ph is not None and rec.phash is None: rec.phash = ph updated += 1 session.commit() last_id = rows[-1].id return updated def _verify_one(path: Path, expected_sha: str, mime: str, sha_fn) -> str: """Compute the integrity verdict for one file. Status precedence: failed_verification (can't run) > corrupt (sha mismatch / decode fails) > ok (passes both). Never raises.""" try: if not path.is_file(): return "failed_verification" try: actual_sha = sha_fn(path) except (OSError, PermissionError): return "failed_verification" if actual_sha != expected_sha: return "corrupt" if mime and mime.startswith("image/"): try: with Image.open(path) as im: im.verify() except Exception: return "corrupt" return "ok" if mime and mime.startswith("video/"): try: proc = subprocess.run( ["ffprobe", "-v", "error", "-i", str(path)], capture_output=True, timeout=FFPROBE_TIMEOUT_SECONDS, ) except FileNotFoundError: # ffprobe binary missing — environment problem, not file. return "failed_verification" except subprocess.TimeoutExpired: return "corrupt" return "ok" if proc.returncode == 0 else "corrupt" # Unknown mime — sha matched already; trust that. return "ok" except Exception as exc: log.warning("verify_integrity unexpected error for %s: %s", path, exc) return "failed_verification" @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 (always — the column reflects the most recent verdict). Keyset-paginated, fail-soft per row, idempotent. Returns the total count verified.""" from ..services.importer import _sha256_of # reuse the importer's helper SessionLocal = _sync_session_factory() total = 0 counts = {"ok": 0, "corrupt": 0, "failed_verification": 0} last_id = 0 with SessionLocal() as session: while True: rows = session.execute( select(ImageRecord) .where(ImageRecord.id > last_id) .order_by(ImageRecord.id.asc()) .limit(VERIFY_PAGE) ).scalars().all() if not rows: break for rec in rows: rec.integrity_status = _verify_one( Path(rec.path), rec.sha256, rec.mime, _sha256_of ) counts[rec.integrity_status] = ( counts.get(rec.integrity_status, 0) + 1 ) total += 1 session.commit() last_id = rows[-1].id log.info("verify_integrity verdicts: %s (total %d)", counts, total) return total @celery.task(name="backend.app.tasks.maintenance.recover_stalled_download_events") def recover_stalled_download_events() -> int: """Recover DownloadEvent rows stuck pending/running past the worker hard kill. The scan tick (scheduler_service.select_due_sources → tasks.scan._tick_due_sources_async) inserts DownloadEvent(status='pending') and fires download_source.delay(). If that task dies before finalizing the event — worker OOM/SIGKILL, lost task, or a gallery-dl that didn't unwind on the 1500s hard time_limit — the event stays in-flight forever. The next tick then skips that source because of the in-flight guard (scan.py:168) and Source.last_checked_at never updates; the operator sees "last check never" in the Subscriptions health column, permanently. This sweep flips matching events to 'error', stamps each affected Source's last_checked_at + last_error and bumps consecutive_failures (once per source, not per event — backoff is exponential on that count so an N-event bump would inflate the next interval by 2^N for no reason). The source becomes re-queueable on the next tick and the health dot goes amber. Operator-confirmed 2026-05-29 (43-row strand pile in production). """ SessionLocal = _sync_session_factory() now = datetime.now(UTC) cutoff = now - timedelta(minutes=DOWNLOAD_STALL_THRESHOLD_MINUTES) msg = "stranded by recovery sweep (no terminal status after time_limit)" with SessionLocal() as session: # UPDATE...RETURNING the source_ids in one round trip — keeps us off # the psycopg 65535-param ceiling that SELECT-then-UPDATE-WHERE-IN # would hit on a large strand pile. result = session.execute( update(DownloadEvent) .where(DownloadEvent.status.in_(["pending", "running"])) .where(DownloadEvent.started_at < cutoff) .values(status="error", finished_at=now, error=msg) .returning(DownloadEvent.source_id) ) returned = result.all() if not returned: session.commit() return 0 events_recovered = len(returned) source_ids = list({row.source_id for row in returned}) session.execute( update(Source) .where(Source.id.in_(source_ids)) .values( consecutive_failures=Source.consecutive_failures + 1, last_error=msg, last_checked_at=now, ) ) session.commit() log.info( "recover_stalled_download_events: recovered %d events across %d sources", events_recovered, len(source_ids), ) 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) db_cutoff = now - timedelta(minutes=BACKUP_DB_STALL_THRESHOLD_MINUTES) slow_cutoff = now - timedelta(minutes=BACKUP_STALL_THRESHOLD_MINUTES) msg = "stranded by recovery sweep (no terminal status within the stall window)" with SessionLocal() as session: result = session.execute( update(BackupRun) .where(BackupRun.status.in_(["running", "restoring"])) # db backups/restores are fast (40-min window); images run hours (7h). .where( or_( and_(BackupRun.kind == "db", BackupRun.started_at < db_cutoff), and_(BackupRun.kind != "db", BackupRun.started_at < slow_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.) Measures staleness from last_progress_at (alembic 0039), NOT started_at: a chunked scan stays 'running' across many re-enqueued chunks and can legitimately run for hours on a big library — only flag one that hasn't made progress in the threshold window (a dead chunk that never re-enqueued). Falls back to started_at for pre-0039 / never-ticked rows. """ SessionLocal = _sync_session_factory() now = datetime.now(UTC) cutoff = now - timedelta(minutes=LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES) msg = ( f"stranded by recovery sweep (no progress for " f"{LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES} min)" ) with SessionLocal() as session: result = session.execute( update(LibraryAuditRun) .where(LibraryAuditRun.status == "running") .where( func.coalesce( LibraryAuditRun.last_progress_at, 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_tag_eval_runs") def recover_stalled_tag_eval_runs() -> int: """Flip TagEvalRun rows stuck in 'running' past the stall threshold to 'error', and prune old runs to the last TAG_EVAL_KEEP_RUNS (retention, rule 89). Runs every 5 min on the maintenance lane; no-op when idle.""" SessionLocal = _sync_session_factory() now = datetime.now(UTC) cutoff = now - timedelta(minutes=TAG_EVAL_STALL_THRESHOLD_MINUTES) with SessionLocal() as session: result = session.execute( update(TagEvalRun) .where(TagEvalRun.status == "running") .where( func.coalesce(TagEvalRun.last_progress_at, TagEvalRun.started_at) < cutoff ) .values( status="error", finished_at=now, error=( f"stranded by recovery sweep (no progress for " f"{TAG_EVAL_STALL_THRESHOLD_MINUTES} min)" ), ) ) # Retention: keep only the most recent N runs. keep = session.execute( select(TagEvalRun.id).order_by(TagEvalRun.id.desc()) .limit(TAG_EVAL_KEEP_RUNS) ).scalars().all() if keep: session.execute( delete(TagEvalRun).where(TagEvalRun.id.not_in(keep)) ) session.commit() recovered = result.rowcount or 0 if recovered: log.info("recover_stalled_tag_eval_runs: recovered %d rows", recovered) return recovered @celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_training_runs") def recover_stalled_head_training_runs() -> int: """Flip HeadTrainingRun rows stuck in 'running' past the stall threshold to 'error', and prune old runs to the last HEAD_TRAINING_KEEP_RUNS (retention, rule 89). Runs every 5 min on the maintenance lane; no-op when idle.""" SessionLocal = _sync_session_factory() now = datetime.now(UTC) cutoff = now - timedelta(minutes=HEAD_TRAINING_STALL_THRESHOLD_MINUTES) with SessionLocal() as session: result = session.execute( update(HeadTrainingRun) .where(HeadTrainingRun.status == "running") .where( func.coalesce( HeadTrainingRun.last_progress_at, HeadTrainingRun.started_at ) < cutoff ) .values( status="error", finished_at=now, error=( f"stranded by recovery sweep (no progress for " f"{HEAD_TRAINING_STALL_THRESHOLD_MINUTES} min)" ), ) ) keep = session.execute( select(HeadTrainingRun.id).order_by(HeadTrainingRun.id.desc()) .limit(HEAD_TRAINING_KEEP_RUNS) ).scalars().all() if keep: session.execute( delete(HeadTrainingRun).where(HeadTrainingRun.id.not_in(keep)) ) session.commit() recovered = result.rowcount or 0 if recovered: log.info( "recover_stalled_head_training_runs: recovered %d rows", recovered ) return recovered @celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_auto_apply_runs") def recover_stalled_head_auto_apply_runs() -> int: """Flip stalled HeadAutoApplyRun 'running' rows to 'error' + prune to the last HEAD_AUTO_APPLY_KEEP_RUNS (retention, rule 89). 5-min maintenance lane.""" SessionLocal = _sync_session_factory() now = datetime.now(UTC) cutoff = now - timedelta(minutes=HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES) with SessionLocal() as session: result = session.execute( update(HeadAutoApplyRun) .where(HeadAutoApplyRun.status == "running") .where( func.coalesce( HeadAutoApplyRun.last_progress_at, HeadAutoApplyRun.started_at ) < cutoff ) .values( status="error", finished_at=now, error=( f"stranded by recovery sweep (no progress for " f"{HEAD_AUTO_APPLY_STALL_THRESHOLD_MINUTES} min)" ), ) ) keep = session.execute( select(HeadAutoApplyRun.id).order_by(HeadAutoApplyRun.id.desc()) .limit(HEAD_AUTO_APPLY_KEEP_RUNS) ).scalars().all() if keep: session.execute( delete(HeadAutoApplyRun).where(HeadAutoApplyRun.id.not_in(keep)) ) session.commit() recovered = result.rowcount or 0 if recovered: log.info( "recover_stalled_head_auto_apply_runs: recovered %d rows", recovered ) return recovered # Keep ~6 months of daily head-metric snapshots (enough to see tuning trends). HEAD_METRICS_SNAPSHOT_RETENTION_DAYS = 180 @celery.task(name="backend.app.tasks.maintenance.snapshot_head_metrics") def snapshot_head_metrics() -> int: """Daily per-concept observability point (#114): record each head-bearing concept's auto-applied volume, cumulative misfires/under-fires, and the head's measured quality — the time-series the operator tunes from. Prunes points older than the retention window.""" from ..models import ( HeadMetric, HeadMetricsSnapshot, Tag, TagHead, ) from ..models.tag import image_tag SessionLocal = _sync_session_factory() now = datetime.now(UTC) with SessionLocal() as session: heads = { r.tag_id: r for r in session.execute( select( TagHead.tag_id, TagHead.ap, TagHead.precision_cv, TagHead.recall, TagHead.n_pos, ) ) } metrics = { r.tag_id: r for r in session.execute( select( HeadMetric.tag_id, HeadMetric.n_misfires, HeadMetric.n_underfires ) ) } applied = dict( session.execute( select(image_tag.c.tag_id, func.count()) .where(image_tag.c.source == "head_auto") .group_by(image_tag.c.tag_id) ) ) tag_ids = set(heads) | set(metrics) if not tag_ids: return 0 names = dict( session.execute(select(Tag.id, Tag.name).where(Tag.id.in_(tag_ids))) ) for tid in tag_ids: h = heads.get(tid) m = metrics.get(tid) session.add(HeadMetricsSnapshot( tag_id=tid, name=names.get(tid, str(tid)), snapshot_at=now, n_auto_applied=applied.get(tid, 0), n_misfires=m.n_misfires if m else 0, n_underfires=m.n_underfires if m else 0, ap=h.ap if h else None, precision_cv=h.precision_cv if h else None, recall=h.recall if h else None, n_pos=h.n_pos if h else None, )) session.execute( delete(HeadMetricsSnapshot).where( HeadMetricsSnapshot.snapshot_at < now - timedelta(days=HEAD_METRICS_SNAPSHOT_RETENTION_DAYS) ) ) session.commit() return len(tag_ids) @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 retention window. Never touches pending/running rows. Why terminal-only: pending/running rows represent in-flight work whose owning task may still be alive; deleting them would orphan the task. Retention days comes from ImportSettings.download_event_retention_days so the operator can tune without a code change. """ SessionLocal = _sync_session_factory() with SessionLocal() as session: settings = ImportSettings.load_sync(session) retention_days = settings.download_event_retention_days cutoff = datetime.now(UTC) - timedelta(days=retention_days) result = session.execute( delete(DownloadEvent) .where(DownloadEvent.status.in_(["ok", "error", "skipped"])) .where(DownloadEvent.started_at < cutoff) ) session.commit() return result.rowcount or 0 @celery.task(name="backend.app.tasks.maintenance.vacuum_analyze") def vacuum_analyze() -> dict: """Periodic VACUUM (ANALYZE) over the high-churn tables (VACUUM_TABLES) to reclaim dead-tuple bloat and refresh planner statistics. VACUUM cannot run inside a transaction block, so it runs on an AUTOCOMMIT connection. Scheduled weekly; also operator-triggerable from Settings → Maintenance. """ engine = get_sync_engine() done = [] with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn: for table in VACUUM_TABLES: conn.exec_driver_sql(f"VACUUM (ANALYZE) {table}") done.append(table) log.info("vacuum_analyze complete: %s", done) return {"vacuumed": done}