From e77afe82952a7d5f25b19944992efa3c5a81639c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 27 May 2026 23:54:35 -0400 Subject: [PATCH] =?UTF-8?q?feat(import-resilience=20L1):=20poison-pill=20c?= =?UTF-8?q?ircuit=20breaker=20=E2=80=94=20cap=20stuck-task=20re-queues?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Layer 1 of the import-task resilience work (operator-requested 2026-05-28). The recover_interrupted_tasks sweep re-queues rows stuck in 'processing' — correct for a worker crash, but without a cap a row that RELIABLY hard-crashes the worker (OOM/segfault/SIGKILL on a corrupt or oversized input) loops forever: re-queue → crash → re-queue, burning a worker slot every 5 min. A caught exception flips to terminal 'failed' and never enters this loop; only process-killing inputs do. - alembic 0026: import_task.recovery_count (int, default 0) + import_task.refetched (bool, default false — backs Layer 2). - recover_interrupted_tasks now runs a poison-pill UPDATE FIRST: stuck rows whose recovery_count has already reached MAX_RECOVERY_ATTEMPTS-1 are marked 'failed' with a diagnostic ("crashed or stalled the worker N times … likely a corrupt or oversized input … inspect/replace the file, then retry via /api/import/retry-failed") instead of re-queued. The re-queue pass then handles the remaining stuck rows and bumps recovery_count. Shared stuck_predicate (and_/or_) keeps the media-5min / archive-40min split. - MAX_RECOVERY_ATTEMPTS=3 (two recoveries then give up). The failed poison pill surfaces in the existing import-failures view with its file path, directly answering "help me identify them." Test test_recover_interrupted_poison_pill_caps_at_max pins both branches: a row at the cap is failed (not re-enqueued, diagnostic present), a row one short is re-queued + incremented. --- ...26_import_task_recovery_count_refetched.py | 53 ++++++++++++++++ backend/app/models/import_task.py | 18 +++++- backend/app/tasks/maintenance.py | 63 ++++++++++++++++--- tests/test_maintenance.py | 46 ++++++++++++++ 4 files changed, 170 insertions(+), 10 deletions(-) create mode 100644 alembic/versions/0026_import_task_recovery_count_refetched.py 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/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/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 797d6f2..f864873 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -6,7 +6,7 @@ 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 @@ -23,6 +23,15 @@ STUCK_THRESHOLD_MINUTES = 5 # 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 @@ -98,18 +107,54 @@ def recover_interrupted_tasks() -> int: # 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(ImportTask.status == "processing") - .where( - ((ImportTask.task_type != "archive") - & (ImportTask.started_at < media_cutoff)) - | ((ImportTask.task_type == "archive") - & (ImportTask.started_at < archive_cutoff)) - ) + .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) @@ -138,7 +183,7 @@ def recover_interrupted_tasks() -> int: for tid, task_type in stuck: enqueue_import(tid, task_type) - return len(stuck) + orphan_count + return len(stuck) + len(poison_ids) + orphan_count @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks") diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 0488943..0f5164e 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -164,6 +164,52 @@ def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch assert dispatched == [stuck.id] # stuck rows re-enqueue; orphans don't +def test_recover_interrupted_poison_pill_caps_at_max(db_sync, monkeypatch): + """A stuck row that's already been recovered MAX_RECOVERY_ATTEMPTS-1 + times is marked 'failed' (with a diagnostic) instead of re-queued — + the circuit breaker against an input that hard-crashes the worker + every run. Operator-flagged 2026-05-28.""" + from backend.app.tasks import import_file + from backend.app.tasks.maintenance import ( + MAX_RECOVERY_ATTEMPTS, + recover_interrupted_tasks, + ) + dispatched: list[int] = [] + monkeypatch.setattr( + import_file.import_media_file, "delay", dispatched.append + ) + + batch_id = _make_batch(db_sync) + now = datetime.now(UTC) + + # At the cap already (recovered MAX-1 times) → fail, don't re-queue. + poison = ImportTask( + batch_id=batch_id, source_path="/import/poison.jpg", task_type="media", + status="processing", started_at=now - timedelta(hours=2), + recovery_count=MAX_RECOVERY_ATTEMPTS - 1, + ) + # One recovery short of the cap → re-queue + increment. + recoverable = ImportTask( + batch_id=batch_id, source_path="/import/ok.jpg", task_type="media", + status="processing", started_at=now - timedelta(hours=2), + recovery_count=MAX_RECOVERY_ATTEMPTS - 2, + ) + db_sync.add_all([poison, recoverable]) + db_sync.commit() + + touched = recover_interrupted_tasks.apply().get() + assert touched == 2 # one failed + one re-queued + + db_sync.refresh(poison) + db_sync.refresh(recoverable) + assert poison.status == "failed" + assert "corrupt or" in (poison.error or "") + assert recoverable.status == "queued" + assert recoverable.recovery_count == MAX_RECOVERY_ATTEMPTS - 1 + # Only the recoverable row re-enqueues; the poison pill does not. + assert dispatched == [recoverable.id] + + def test_cleanup_old_deletes_finished_old(db_sync): batch_id = _make_batch(db_sync) now = datetime.now(UTC)