diff --git a/alembic/versions/0026_import_task_recovery_count_refetched.py b/alembic/versions/0026_import_task_recovery_count_refetched.py new file mode 100644 index 0000000..ccbc3da --- /dev/null +++ b/alembic/versions/0026_import_task_recovery_count_refetched.py @@ -0,0 +1,53 @@ +"""import_task.recovery_count + refetched — poison-pill circuit breaker + +Revision ID: 0026 +Revises: 0025 +Create Date: 2026-05-28 + +Backs the import-task resilience work (operator-flagged 2026-05-28): + +- recovery_count: how many times recover_interrupted_tasks has + re-queued this row from a stuck 'processing' state. A row that + hard-crashes the worker (OOM / segfault on a corrupt or oversized + input) leaves no terminal flip, so the sweep re-queues it — and + without a cap it would loop forever, re-crashing the worker each + time. After MAX_RECOVERY_ATTEMPTS the sweep marks it 'failed' with a + diagnostic instead. + +- refetched: whether a one-shot re-download has already been attempted + for this task's file. Bounds the Layer-2 re-fetch remediation to a + single attempt so source-side corruption doesn't loop. + +Both default to 0 / false; additive, no backfill needed. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0026" +down_revision: Union[str, None] = "0025" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "import_task", + sa.Column( + "recovery_count", sa.Integer(), nullable=False, + server_default="0", + ), + ) + op.add_column( + "import_task", + sa.Column( + "refetched", sa.Boolean(), nullable=False, + server_default=sa.false(), + ), + ) + + +def downgrade() -> None: + op.drop_column("import_task", "refetched") + op.drop_column("import_task", "recovery_count") diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py index 02f65d2..7ad3fa7 100644 --- a/backend/app/api/import_admin.py +++ b/backend/app/api/import_admin.py @@ -114,18 +114,55 @@ async def retry_failed(): status="queued", error=None, started_at=None, finished_at=None, ) - .returning(ImportTask.id) + .returning(ImportTask.id, ImportTask.task_type) ) - failed_ids = [row[0] for row in result.all()] - if not failed_ids: + failed = result.all() + if not failed: return jsonify({"retried": 0}) await session.commit() - from ..tasks.import_file import import_media_file - for tid in failed_ids: - import_media_file.delay(tid) + from ..tasks.import_file import enqueue_import + for tid, task_type in failed: + enqueue_import(tid, task_type) - return jsonify({"retried": len(failed_ids)}) + return jsonify({"retried": len(failed)}) + + +@import_admin_bp.route("/tasks//refetch", methods=["POST"]) +async def refetch_task(task_id: int): + """Layer-2 one-shot re-download: delete the (corrupt) file behind a + failed import task and re-run its source's downloader to fetch a + fresh copy. Only works for files that resolve to an enabled, + real-URL subscription Source; filesystem-only imports return + no_source. + + Returns one of: refetch_queued (+source_id) / no_source / + already_refetched / not_found / not_failed. + """ + async with get_session() as session: + result = await session.run_sync(_refetch_task_sync, task_id) + if result["status"] == "not_found": + return jsonify(result), 404 + if result["status"] == "not_failed": + return jsonify(result), 400 + return jsonify(result) + + +def _refetch_task_sync(session, task_id: int) -> dict: + from pathlib import Path + + from ..models import ImportSettings + from ..services.refetch_service import attempt_refetch + + task = session.get(ImportTask, task_id) + if task is None: + return {"status": "not_found"} + if task.status != "failed": + return {"status": "not_failed"} + settings = session.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + return attempt_refetch(session, task, Path(settings.import_scan_path)) @import_admin_bp.route("/clear-stuck", methods=["POST"]) diff --git a/backend/app/models/import_task.py b/backend/app/models/import_task.py index da3d4ec..3c947c1 100644 --- a/backend/app/models/import_task.py +++ b/backend/app/models/import_task.py @@ -8,7 +8,16 @@ been processing longer than the stuck-task threshold. from datetime import datetime -from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, Text, func +from sqlalchemy import ( + BigInteger, + Boolean, + DateTime, + ForeignKey, + Integer, + String, + Text, + func, +) from sqlalchemy.orm import Mapped, mapped_column, relationship from .base import Base @@ -26,6 +35,13 @@ class ImportTask(Base): task_type: Mapped[str] = mapped_column(String(16), nullable=False) # media|archive status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True) + # Poison-pill circuit breaker (alembic 0026). recovery_count tracks + # how many times the stuck-task sweep has re-queued this row; after + # the cap it's failed with a diagnostic instead of looping. refetched + # bounds the one-shot re-download remediation to a single attempt. + recovery_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + refetched: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + result_image_id: Mapped[int | None] = mapped_column( ForeignKey("image_record.id", ondelete="SET NULL"), nullable=True ) diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index e2b94eb..0e366d9 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -30,6 +30,7 @@ from ..models import ( PostAttachment, Source, ) +from ..utils import safe_probe from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name from ..utils.phash import compute_phash, find_similar from ..utils.sidecar import find_sidecar, parse_sidecar @@ -407,6 +408,29 @@ class Importer: return ImportResult(status="attached") def _import_archive(self, source: Path) -> ImportResult: + # Layer-3 isolation: bomb-size guard + integrity test in a + # spawned child BEFORE extracting in this process. A + # decompression bomb or a native-lib crash on a malformed + # archive is contained to the child; we reject the file cleanly + # instead of OOMing/segfaulting the import worker. extract_archive + # is already fail-soft for plain exceptions, so this only adds + # the hard-crash protection. + probe = safe_probe.probe_archive(source) + if not probe.ok: + if probe.crashed: + return ImportResult( + status="failed", + error=f"archive probe crashed/timed out: {probe.reason}", + ) + # Clean rejection (bomb cap exceeded, integrity mismatch): + # still preserve the archive file itself as an attachment so + # nothing silently vanishes, matching extract_archive's + # fail-soft contract. + artist = self._resolve_artist(source) + post = self._post_for_sidecar(source, artist) + self._capture_attachment(source, post=post, artist=artist, resolved=True) + return ImportResult(status="attached") + artist = self._resolve_artist(source) post = self._post_for_sidecar(source, artist) member_ids: list[int] = [] @@ -446,7 +470,25 @@ class Importer: # Compute file dimensions (images only) and apply filters. width = height = None has_alpha = False - if not is_video(source): + if is_video(source): + # Layer-3 isolation: validate the container via ffprobe (a + # separate process) before the rest of the pipeline touches + # it. A corrupt video that would crash a decoder is rejected + # cleanly here, and we capture width/height for free (the + # importer didn't previously record video dimensions). + probe = safe_probe.probe_video(source) + if not probe.ok: + if probe.crashed: + return ImportResult( + status="failed", + error=f"video probe crashed/timed out: {probe.reason}", + ) + return ImportResult( + status="skipped", skip_reason=SkipReason.invalid_image, + error=probe.reason, + ) + width, height = probe.width, probe.height + else: try: with Image.open(source) as im: im.verify() diff --git a/backend/app/services/refetch_service.py b/backend/app/services/refetch_service.py new file mode 100644 index 0000000..41832bc --- /dev/null +++ b/backend/app/services/refetch_service.py @@ -0,0 +1,106 @@ +"""Layer-2 one-shot re-download remediation for corrupt imported files. + +When an import fails on a file that came from a known, pollable +subscription Source, deleting the bad copy and re-running the source's +downloader can fetch a fresh, unblemished copy. This only helps when: + + - the corruption is in transit / on disk (not at the source), AND + - the file resolves to an ENABLED Source with a real feed URL + (a `sidecar::` synthetic anchor is not pollable), + AND + - we haven't already re-fetched this task once (bounded by + ImportTask.refetched so source-side corruption can't loop). + +Filesystem-only imports with no resolvable Source return 'no_source' — +the operator's only remediation there is to replace the file on disk. + +Operator-requested 2026-05-28 (Layer 2). +""" + +import json +import logging +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..models import Artist, ImportTask, Source +from ..utils.paths import derive_top_level_artist +from ..utils.sidecar import find_sidecar, parse_sidecar +from ..utils.slug import slugify + +log = logging.getLogger(__name__) + + +def resolve_refetch_source( + session: Session, source_path: str, import_root: Path, +) -> Source | None: + """Find an enabled, real-URL Source for the file's (artist, platform), + or None when nothing re-pollable resolves.""" + path = Path(source_path) + sc = find_sidecar(path) + if sc is None: + return None + try: + data = json.loads(sc.read_text("utf-8")) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(data, dict): + return None + sd = parse_sidecar(data) + if not sd.platform: + return None + artist_name = derive_top_level_artist(path, import_root) + if not artist_name: + return None + artist = session.execute( + select(Artist).where(Artist.slug == slugify(artist_name)) + ).scalar_one_or_none() + if artist is None: + return None + src = session.execute( + select(Source) + .where( + Source.artist_id == artist.id, + Source.platform == sd.platform, + Source.enabled.is_(True), + ) + .order_by(Source.id.asc()) + ).scalars().first() + if src is None: + return None + if (src.url or "").startswith("sidecar:"): + return None # synthetic anchor — not a pollable feed + return src + + +def attempt_refetch( + session: Session, task: ImportTask, import_root: Path, +) -> dict: + """Delete the corrupt file, mark the task refetched, and trigger ONE + source re-check. Idempotent/bounded: a task already refetched (or + with no resolvable Source) is a no-op. Commits.""" + if task.refetched: + return {"status": "already_refetched"} + src = resolve_refetch_source(session, task.source_path, import_root) + if src is None: + return {"status": "no_source"} + + # Remove the bad copy so gallery-dl (skip_existing) re-fetches it on + # the source re-check instead of skipping the still-present corrupt + # file. + try: + Path(task.source_path).unlink(missing_ok=True) + except OSError as exc: + log.warning("refetch unlink failed for %s: %s", task.source_path, exc) + + task.refetched = True + session.add(task) + session.commit() + + # Lazy import to avoid a tasks→services→tasks import cycle at module + # load. download_source.delay() is sync-safe in any context. + from ..tasks.download import download_source + + download_source.delay(src.id) + return {"status": "refetch_queued", "source_id": src.id} diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py index e0752e8..4a124d1 100644 --- a/backend/app/tasks/import_file.py +++ b/backend/app/tasks/import_file.py @@ -64,30 +64,13 @@ def _mark_failed(session, task, error_msg: str) -> None: pass -@celery.task( - name="backend.app.tasks.import_file.import_media_file", - bind=True, - autoretry_for=(OperationalError, DBAPIError, OSError), - retry_backoff=5, - retry_backoff_max=60, - retry_jitter=True, - max_retries=3, - soft_time_limit=300, - time_limit=360, -) -def import_media_file(self, import_task_id: int) -> dict: - """Returns a dict so the eager-mode tests can assert without DB. - - Decorator notes: - - autoretry_for: transient DB / filesystem errors retry with - exponential backoff (5s base, jitter, max 3 attempts). On final - give-up the task raises and acks_late=True (set globally on the - Celery app) does NOT redeliver — the recovery sweep catches the - row instead. - - soft_time_limit (300s) raises SoftTimeLimitExceeded in this - process so the task can mark its row failed before being killed. - - time_limit (360s) is the hard cap; SIGKILL if the soft signal - was swallowed. +def _run_import_task(import_task_id: int) -> dict: + """Shared body for import_media_file + import_archive_file. The two + tasks differ ONLY in their Celery time limits (a single media file + is sub-second; an archive runs the full per-member pipeline inline + for every member and can take many minutes). Both flip the row to + 'processing', dispatch to `_do_import`, and honor the + flip-to-terminal resilience contract. """ SessionLocal = _sync_session_factory() with SessionLocal() as session: @@ -103,19 +86,85 @@ def import_media_file(self, import_task_id: int) -> dict: try: return _do_import(session, task, import_task_id) except SoftTimeLimitExceeded: - _mark_failed(session, task, "soft_time_limit exceeded (>300s)") + _mark_failed(session, task, "soft_time_limit exceeded") raise except (OperationalError, DBAPIError, OSError): # Retryable per the decorator; do NOT mark failed (let # autoretry have a clean go at it). If autoretry exhausts, # the row stays 'processing' and the maintenance sweep - # flips it within 5 min. + # flips it. raise except Exception as exc: # noqa: BLE001 — pipeline crash, mark + re-raise _mark_failed(session, task, f"{type(exc).__name__}: {exc}") raise +@celery.task( + name="backend.app.tasks.import_file.import_media_file", + bind=True, + autoretry_for=(OperationalError, DBAPIError, OSError), + retry_backoff=5, + retry_backoff_max=60, + retry_jitter=True, + max_retries=3, + soft_time_limit=300, + time_limit=360, +) +def import_media_file(self, import_task_id: int) -> dict: + """Import ONE media file (or non-media → PostAttachment). Sub-second + for the common case; the tight 5-min soft limit keeps a genuinely + stuck single-file import detectable fast. + + Decorator notes: + - autoretry_for: transient DB / filesystem errors retry with + exponential backoff (5s base, jitter, max 3 attempts). On final + give-up the task raises and acks_late=True (set globally on the + Celery app) does NOT redeliver — the recovery sweep catches the + row instead. + - soft_time_limit (300s) raises SoftTimeLimitExceeded in-process + so the task can mark its row failed before being killed. + - time_limit (360s) is the hard SIGKILL cap. + """ + return _run_import_task(import_task_id) + + +@celery.task( + name="backend.app.tasks.import_file.import_archive_file", + bind=True, + autoretry_for=(OperationalError, DBAPIError, OSError), + retry_backoff=5, + retry_backoff_max=60, + retry_jitter=True, + max_retries=3, + # Archives run the full per-member pipeline (sha256 + pHash + dedup + # query + copy + provenance) for EVERY media member inline, under a + # single task budget. A multi-hundred-member archive blows the + # 5-min media limit. soft=30min / hard=35min sizes for a large + # archive. Operator-flagged 2026-05-28 (target 1645019 hit the old + # shared 300s soft limit). The recovery sweep gives this task its + # own 40-min threshold via maintenance.TASK_STUCK_THRESHOLD_MINUTES + # so it isn't preempted while legitimately grinding through members. + soft_time_limit=1800, + time_limit=2100, +) +def import_archive_file(self, import_task_id: int) -> dict: + """Import an archive: extract + run the per-member media pipeline for + every member inline, then preserve the archive as a PostAttachment. + Same body as import_media_file (dispatch is by file kind inside + Importer.import_one); split out purely for the larger time budget.""" + return _run_import_task(import_task_id) + + +def enqueue_import(task_id: int, task_type: str) -> None: + """Route an ImportTask to the right Celery task by its task_type. + Single source of truth for the media-vs-archive dispatch so the + scan, retry, and recovery-requeue paths stay in sync.""" + if task_type == "archive": + import_archive_file.delay(task_id) + else: + import_media_file.delay(task_id) + + def _do_import(session, task, import_task_id: int) -> dict: """Actual work, called from inside the resilience wrapper.""" settings = session.execute( diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index df854a3..4584d01 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1,12 +1,13 @@ """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 delete, select, update +from sqlalchemy import and_, delete, or_, select, update from ..celery_app import celery from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun @@ -16,6 +17,22 @@ from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) 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 OLD_TASK_DAYS = 7 PHASH_PAGE = 500 @@ -24,17 +41,40 @@ FFPROBE_TIMEOUT_SECONDS = 10 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days +# 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, +} +TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = { + "backend.app.tasks.import_file.import_archive_file": 40, +} + @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. + 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 @@ -51,7 +91,8 @@ def recover_interrupted_tasks() -> int: """ SessionLocal = _sync_session_factory() now = datetime.now(UTC) - processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + 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 @@ -59,20 +100,67 @@ def recover_interrupted_tasks() -> int: # 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( + # 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(ImportTask.status == "processing") - .where(ImportTask.started_at < processing_cutoff) + .where(stuck_predicate) + .where(ImportTask.recovery_count >= MAX_RECOVERY_ATTEMPTS - 1) .values( - status="queued", - started_at=None, - error="recovered from stuck state", + 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) ) - stuck_ids = [row[0] for row in stuck_result.all()] + 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) @@ -91,12 +179,34 @@ def recover_interrupted_tasks() -> int: session.commit() - if stuck_ids: - from .import_file import import_media_file - for tid in stuck_ids: - import_media_file.delay(tid) + if stuck: + from .import_file import enqueue_import + for tid, task_type in stuck: + enqueue_import(tid, task_type) - return len(stuck_ids) + orphan_count + # 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") @@ -121,18 +231,29 @@ def cleanup_old_tasks() -> int: @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. + """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). Shares the 5-min threshold - with recover_interrupted_tasks for consistency. + 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() - cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES) - with SessionLocal() as session: - result = session.execute( + 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) @@ -140,14 +261,42 @@ def recover_stalled_task_runs() -> int: status="error", error_type="RecoverySweep", error_message=( - f"no completion signal received within " - f"{STUCK_THRESHOLD_MINUTES} min" + f"no completion signal received within {minutes} min" ), - finished_at=datetime.now(UTC), + finished_at=now, ) ) + 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 result.rowcount or 0 + return total @celery.task(name="backend.app.tasks.maintenance.prune_task_runs") diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 75de41d..b114eec 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -31,8 +31,15 @@ def _is_video(path: Path) -> bool: retry_backoff_max=60, retry_jitter=True, max_retries=3, - soft_time_limit=300, - time_limit=420, + # Sized for the video branch: sample 10 frames, run tagger + + # embedder on each (≈20 GPU ops vs 2 for an image). A loaded + # ml-worker can take 5-10 min on a long video; bumped from + # 5min/7min on 2026-05-28 after operator-flagged image 6288 (a + # .mp4) hit the recovery sweep at 5 min while still legitimately + # processing. Image runs return in seconds; the bump doesn't + # affect their UX. + soft_time_limit=900, # 15 min + time_limit=1200, # 20 min hard ) def tag_and_embed(self, image_id: int) -> dict: """Run Camie + SigLIP on one image; store predictions + embedding; @@ -64,6 +71,18 @@ def tag_and_embed(self, image_id: int) -> dict: embedder = get_embedder() if _is_video(src): + # Layer-3 isolation: ffprobe (a separate process) validates + # the container before we burn ~20 GPU ops sampling frames + # from it. A corrupt video that would crash the frame + # decoder is rejected cleanly here instead of taking down + # the ml-worker. Operator-flagged 2026-05-28. + from ..utils import safe_probe + vprobe = safe_probe.probe_video(src) + if not vprobe.ok: + return { + "status": "bad_video", "image_id": image_id, + "reason": vprobe.reason, + } frames = _sample_video_frames( src, int(os.environ.get("VIDEO_ML_FRAMES", "10")) ) diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 2e5b612..0451e49 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -17,6 +17,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn from ..celery_app import celery from ..config import get_config from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask +from ..services.archive_extractor import is_archive from ..services.scheduler_service import select_due_sources from ._sync_engine import sync_session_factory as _sync_session_factory @@ -96,7 +97,9 @@ def scan_directory(self, triggered_by: str = "manual", task = ImportTask( batch_id=batch_id, source_path=entry_str, - task_type="media", + # Archives route to import_archive_file (larger time + # budget) — they run the per-member pipeline inline. + task_type="archive" if is_archive(entry) else "media", status="pending", size_bytes=size, ) @@ -115,15 +118,16 @@ def scan_directory(self, triggered_by: str = "manual", batch.finished_at = datetime.now(UTC) session.commit() - # Now enqueue import_media_file for each pending task. + # Now enqueue each pending task on the right Celery task + # (media vs archive) via the shared router. + from .import_file import enqueue_import + for task in session.execute( select(ImportTask).where(ImportTask.batch_id == batch_id) ).scalars(): task.status = "queued" session.add(task) - from .import_file import import_media_file - - import_media_file.delay(task.id) + enqueue_import(task.id, task.task_type) session.commit() if mode == "deep": diff --git a/backend/app/utils/safe_probe.py b/backend/app/utils/safe_probe.py new file mode 100644 index 0000000..02c387c --- /dev/null +++ b/backend/app/utils/safe_probe.py @@ -0,0 +1,170 @@ +"""Subprocess-isolated media probes (Layer 3 of import resilience). + +A malformed video or archive can hard-crash the worker process — a +decoder OOM, a native-lib segfault, or a decompression bomb. A hard +crash leaves no terminal flip, so the recovery sweep re-queues the row +and it crashes again: a poison-pill loop (the Layer-1 cap is the +backstop, but isolating the crash is better — the file gets a clean +terminal failure and the worker never dies). + +These probes run the risky read in a way that contains the blast: + +- Video: `ffprobe` is a separate binary, so a crash decoding the + container kills only ffprobe (non-zero exit), never the worker. Also + returns width/height, which the importer didn't previously capture + for videos. +- Archive: an uncompressed-size guard (catches decompression bombs + before they OOM anything) plus an integrity test in a spawned child + (catches native-lib crashes on a malformed archive). A child segfault + / OOM shows up as a non-zero exit code, not a dead worker. + +Images are intentionally NOT probed here: Pillow raises (it doesn't +segfault) on the realistic corrupt-image cases, the importer already +catches that as an invalid_image skip, and a subprocess per image would +wreck deep-scan throughput on a large library. Add an image branch only +if a real image-induced worker crash is ever observed. + +Operator-requested 2026-05-28 (Layer 3). +""" + +import json +import multiprocessing as mp +import subprocess +from dataclasses import dataclass +from pathlib import Path + +VIDEO_PROBE_TIMEOUT_SECONDS = 60 +ARCHIVE_PROBE_TIMEOUT_SECONDS = 120 +# Refuse archives whose total UNCOMPRESSED size exceeds this — the +# classic decompression-bomb guard (a 4 GB cap comfortably clears real +# art-pack archives while stopping a few-KB zip that expands to TB). +MAX_ARCHIVE_UNCOMPRESSED_BYTES = 4 * 1024 * 1024 * 1024 + + +@dataclass(frozen=True) +class ProbeResult: + ok: bool + # crashed=True means the probe HARD-FAILED (subprocess killed by a + # signal, OOM, or timeout) — the poison-pill signature. crashed=False + # with ok=False means a clean rejection (corrupt-but-handled, + # bomb-size-exceeded, integrity mismatch). Callers map crashed → a + # terminal 'failed', clean → a 'skipped'/'failed' of their choosing. + crashed: bool = False + reason: str | None = None + width: int | None = None + height: int | None = None + + +def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) -> ProbeResult: + """Validate a video container + first video stream via ffprobe.""" + try: + out = subprocess.run( + [ + "ffprobe", "-v", "error", + "-select_streams", "v:0", + "-show_entries", "stream=width,height", + "-of", "json", str(path), + ], + capture_output=True, text=True, timeout=timeout, + ) + except subprocess.TimeoutExpired: + return ProbeResult(ok=False, crashed=True, reason="ffprobe timed out") + except OSError as exc: + # ffprobe missing / not executable — environmental, not the + # file's fault. Treat as a clean non-crash failure so the import + # path can decide (it currently proceeds without dims). + return ProbeResult(ok=False, crashed=False, reason=f"ffprobe unavailable: {exc}") + if out.returncode != 0: + return ProbeResult( + ok=False, crashed=False, + reason=f"ffprobe rejected the file: {out.stderr.strip()[:200]}", + ) + try: + streams = (json.loads(out.stdout) or {}).get("streams") or [] + except json.JSONDecodeError as exc: + return ProbeResult(ok=False, crashed=False, reason=f"ffprobe output parse failed: {exc}") + if not streams: + return ProbeResult(ok=False, crashed=False, reason="no decodable video stream") + return ProbeResult( + ok=True, width=streams[0].get("width"), height=streams[0].get("height"), + ) + + +def probe_archive(path: Path, *, timeout: float = ARCHIVE_PROBE_TIMEOUT_SECONDS) -> ProbeResult: + """Bomb-size guard + isolated integrity test for an archive.""" + ctx = mp.get_context("spawn") + q = ctx.Queue() + proc = ctx.Process(target=_archive_probe_target, args=(str(path), q)) + proc.start() + proc.join(timeout) + if proc.is_alive(): + proc.terminate() + proc.join(5) + return ProbeResult(ok=False, crashed=True, reason="archive probe timed out") + if proc.exitcode != 0: + # Negative exitcode = killed by signal (segfault); positive = + # the child os._exit'd or was OOM-killed. Either way the file + # hard-crashed the probe — the poison-pill signature. + return ProbeResult( + ok=False, crashed=True, + reason=f"archive probe crashed (exit {proc.exitcode})", + ) + try: + outcome = q.get(timeout=5) + except Exception: # noqa: BLE001 — empty queue / broken pipe + return ProbeResult(ok=False, crashed=True, reason="archive probe produced no result") + status, detail = outcome + if status == "ok": + return ProbeResult(ok=True) + return ProbeResult(ok=False, crashed=False, reason=detail) + + +def _archive_probe_target(path_str: str, q) -> None: + """Runs in the spawned child. Reads member sizes (bomb guard) then + runs the format's integrity test. Puts ('ok', None) or + ('error', reason). A crash/OOM here never reaches the queue — the + parent reads the non-zero exit code instead.""" + path = Path(path_str) + ext = path.suffix.lower() + try: + total, test_bad = _inspect_archive(path, ext) + except Exception as exc: # noqa: BLE001 — clean rejection + q.put(("error", f"{type(exc).__name__}: {exc}")) + return + if total is not None and total > MAX_ARCHIVE_UNCOMPRESSED_BYTES: + gib = total / (1024 ** 3) + q.put(("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap")) + return + if test_bad is not None: + q.put(("error", f"integrity test failed at member {test_bad!r}")) + return + q.put(("ok", None)) + + +def _inspect_archive(path: Path, ext: str): + """Return (total_uncompressed_bytes | None, first_bad_member | None) + for the archive. Format-specific; raises on a structurally-broken + container (caught by the child as a clean rejection).""" + if ext in (".zip", ".cbz"): + import zipfile + + with zipfile.ZipFile(path) as zf: + total = sum(zi.file_size for zi in zf.infolist()) + return total, zf.testzip() + if ext == ".rar": + import rarfile + + with rarfile.RarFile(path) as rf: + total = sum(getattr(ri, "file_size", 0) for ri in rf.infolist()) + rf.testrar() + return total, None + if ext == ".7z": + import py7zr + + with py7zr.SevenZipFile(path, "r") as zf: + info = zf.archiveinfo() + total = getattr(info, "uncompressed", None) + ok = zf.test() # True / None when all members pass + return total, (None if ok in (True, None) else "7z test reported corruption") + # Unknown extension — nothing to test; treat as clean. + return None, None diff --git a/frontend/src/components/downloads/DownloadEventRow.vue b/frontend/src/components/downloads/DownloadEventRow.vue index a4360c0..fb2999b 100644 --- a/frontend/src/components/downloads/DownloadEventRow.vue +++ b/frontend/src/components/downloads/DownloadEventRow.vue @@ -1,47 +1,118 @@ diff --git a/frontend/src/components/settings/ImportTaskList.vue b/frontend/src/components/settings/ImportTaskList.vue index 97bbebf..4eadf3c 100644 --- a/frontend/src/components/settings/ImportTaskList.vue +++ b/frontend/src/components/settings/ImportTaskList.vue @@ -51,6 +51,19 @@ title="Click for full error" >{{ shorten(item.error, 60) }} +
Load more @@ -149,9 +162,29 @@ const headers = [ { title: 'Source', key: 'source_path', sortable: false }, { title: 'Size', key: 'size_bytes', sortable: false, width: 90 }, { title: 'Created', key: 'created_at', sortable: false, width: 150 }, - { title: 'Note', key: 'error', sortable: false } + { title: 'Note', key: 'error', sortable: false }, + { title: '', key: 'actions', sortable: false, width: 56 } ] +const refetching = ref(null) +const _REFETCH_MSG = { + refetch_queued: { text: 'Re-fetch queued — re-downloading from source', type: 'success' }, + no_source: { text: 'No re-fetchable source (filesystem import — replace the file manually)', type: 'info' }, + already_refetched: { text: 'Already re-fetched once', type: 'info' }, +} +async function onRefetch(item) { + refetching.value = item.id + try { + const res = await store.refetchTask(item.id) + const msg = _REFETCH_MSG[res.status] || { text: `Re-fetch: ${res.status}`, type: 'info' } + window.__fcToast?.(msg) + } catch (e) { + window.__fcToast?.({ text: `Re-fetch failed: ${e.message}`, type: 'error' }) + } finally { + refetching.value = null + } +} + const hasFailed = computed(() => store.tasks.some(t => t.status === 'failed')) const hasStuck = computed(() => store.tasks.some( t => t.status === 'pending' || t.status === 'queued' || t.status === 'processing' diff --git a/frontend/src/components/subscriptions/DownloadsTab.vue b/frontend/src/components/subscriptions/DownloadsTab.vue index 7ced798..ecaa92b 100644 --- a/frontend/src/components/subscriptions/DownloadsTab.vue +++ b/frontend/src/components/subscriptions/DownloadsTab.vue @@ -20,15 +20,44 @@
-
+

No download events match the current filter.

- +
+
+ + {{ collapsed[g.key] ? 'mdi-chevron-right' : 'mdi-chevron-down' }} + + {{ g.label }} + + {{ g.failedCount }} + + {{ g.items.length }} + {{ g.items.length === 1 ? 'event' : 'events' }} + + +
+
+ +
+
+
Load more @@ -45,7 +74,7 @@