326 lines
13 KiB
Python
326 lines
13 KiB
Python
"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
|
|
|
|
import logging
|
|
import subprocess
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from PIL import Image
|
|
from sqlalchemy import delete, select, update
|
|
|
|
from ..celery_app import celery
|
|
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
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
STUCK_THRESHOLD_MINUTES = 5
|
|
ORPHAN_PENDING_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
|
|
|
|
|
|
@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' > 5 min — worker crash mid-import. Re-queue via
|
|
.delay() and let the import retry. Was 30 min historically;
|
|
tightened 2026-05-24 after operator hit a 2224-row zombie pile.
|
|
import_media_file is sub-second for the vast majority of files and
|
|
capped at the per-task soft_time_limit (5 min), so anything still
|
|
'processing' after that window is a confirmed crash.
|
|
|
|
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)
|
|
processing_cutoff = now - timedelta(minutes=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 ids that flipped so the stuck sweep can still
|
|
# .delay() each one.
|
|
stuck_result = session.execute(
|
|
update(ImportTask)
|
|
.where(ImportTask.status == "processing")
|
|
.where(ImportTask.started_at < processing_cutoff)
|
|
.values(
|
|
status="queued",
|
|
started_at=None,
|
|
error="recovered from stuck state",
|
|
)
|
|
.returning(ImportTask.id)
|
|
)
|
|
stuck_ids = [row[0] for row in stuck_result.all()]
|
|
|
|
orphan_result = session.execute(
|
|
update(ImportTask)
|
|
.where(ImportTask.status.in_(["pending", "queued"]))
|
|
.where(ImportTask.created_at < orphan_cutoff)
|
|
.values(
|
|
status="failed",
|
|
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_ids:
|
|
from .import_file import import_media_file
|
|
for tid in stuck_ids:
|
|
import_media_file.delay(tid)
|
|
|
|
return len(stuck_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' 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
|
|
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")
|
|
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.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 = session.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
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
|