diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py index 8e6a434..dac6a6c 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -59,29 +59,33 @@ TICK_SKIP_VALUE = "exit:20" # Source.backfill_runs_remaining > 0 selects this mode; the longer # timeout below absorbs creators with thousands of posts. # -# 30 seconds shy of Celery's hard `time_limit=1200` on download_source -# (tasks/download.py:33). subprocess.run MUST raise TimeoutExpired -# before Celery SIGKILLs the worker — same rationale as the tick -# default at line 74. The audit (2026-06-02) caught this at 1800, -# guaranteeing SIGKILL on any backfill that ran to its subprocess -# budget: stdout/stderr lost, backfill_runs_remaining never -# decrements, recovery sweep stamps generic "stranded" 30 min later. -# Recreates the exact Knuxy #38275 failure mode the tick 870s default -# was added to prevent. backfill_runs_remaining=3 still gives ~58 -# minutes of cumulative walk across three runs for prolific creators. +# Sits below download_source's Celery soft_time_limit +# (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py) with ~180s of +# headroom for phase-3 persist. subprocess.run MUST raise TimeoutExpired +# before Celery raises SoftTimeLimitExceeded — that exception path +# captures partial stdout/stderr and finalizes the event; the soft-limit +# path (until the 2026-06-03 fix) did not. Audit history: 1800 guaranteed +# SIGKILL against the old hard limit (Knuxy #38275); 1170 was then sized +# "30s shy of the hard limit (1200)" but still EXCEEDED the soft limit +# (900), so SoftTimeLimitExceeded preempted TimeoutExpired and every +# backfill stranded empty (Anduo #39912). Raising the Celery soft/hard +# limits to 1350/1500 (tasks/download.py) is what made 1170 safe. +# backfill_runs_remaining=3 still gives ~58 minutes of cumulative walk +# across three runs for prolific creators. BACKFILL_SKIP_VALUE = True BACKFILL_TIMEOUT_SECONDS = 1170 -# 30 seconds shy of download_source's Celery soft_time_limit (900s, see -# tasks/download.py:32). subprocess.run MUST raise TimeoutExpired before -# Celery raises SoftTimeLimitExceeded — otherwise Celery wins the race, -# SIGKILLs the worker, in-memory stdout/stderr is lost, and the -# DownloadEvent ends up empty-logged with "stranded by recovery sweep" -# 18 minutes later (operator-flagged 2026-05-31, Knuxy event #38275). -# The 30s buffer absorbs scheduler jitter / GC pauses without making -# legitimately-long-running syncs timeout-friendlier. Per-source bumps -# still live in source.config_overrides for legitimately long syncs. +# Sits well below download_source's Celery soft_time_limit +# (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py). subprocess.run MUST +# raise TimeoutExpired before Celery raises SoftTimeLimitExceeded — +# otherwise Celery wins the race, SIGKILLs the worker, in-memory +# stdout/stderr is lost, and the DownloadEvent ends up empty-logged with +# "stranded by recovery sweep" (operator-flagged 2026-05-31, Knuxy event +# #38275; recurred in backfill mode as Anduo #39912). Per-source bumps +# still live in source.config_overrides for legitimately long syncs — +# keep any override below the soft limit, or the soft-limit salvage path +# in tasks/download.py (_finalize_soft_limited) is the only safety net. _DEFAULT_GDL_TIMEOUT_SECONDS = 870 diff --git a/backend/app/tasks/download.py b/backend/app/tasks/download.py index 7cd7c13..bceee3a 100644 --- a/backend/app/tasks/download.py +++ b/backend/app/tasks/download.py @@ -1,12 +1,17 @@ """download_source Celery task — runs DownloadService for one source.""" import asyncio +import logging +from datetime import UTC, datetime from pathlib import Path +from celery.exceptions import SoftTimeLimitExceeded +from sqlalchemy import select from sqlalchemy.exc import DBAPIError, OperationalError +from sqlalchemy.orm import Session as SyncSession from ..celery_app import celery -from ..models import ImportSettings +from ..models import DownloadEvent, ImportSettings, Source from ..services.credential_crypto import CredentialCrypto from ..services.credential_service import CredentialService from ..services.download_service import DownloadService @@ -16,9 +21,79 @@ from ..services.thumbnailer import Thumbnailer from ._async_session import async_session_factory from .import_file import _sync_session_factory +log = logging.getLogger(__name__) + IMAGES_ROOT = Path("/images") _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64" +# Celery time budget for one download_source run. The ceiling that +# governs *clean* teardown is the SOFT limit: it raises a catchable +# SoftTimeLimitExceeded in-process, whereas the HARD limit SIGKILLs the +# worker (no chance to finalize). Both gallery-dl subprocess budgets +# (gallery_dl.py: _DEFAULT_GDL_TIMEOUT_SECONDS=870 tick, +# BACKFILL_TIMEOUT_SECONDS=1170 backfill) MUST sit below the soft limit +# so subprocess.run raises its own TimeoutExpired first — that path +# captures partial stdout/stderr and finalizes the DownloadEvent. soft is +# max-subprocess (1170) + ~180s phase-3 persist headroom; hard is soft + +# 150s SIGKILL backstop. Audit 2026-06-03 (Anduo #39912): the old +# soft=900 sat BELOW the 1170 backfill budget, so SoftTimeLimitExceeded +# preempted TimeoutExpired and the event stranded empty. The recovery +# sweep's DOWNLOAD_STALL_THRESHOLD_MINUTES (30 min) still trails the new +# 25-min hard kill by 5 min, so it stays a true backstop. Invariant +# guarded by test_timeout_ladder_keeps_subprocess_budgets_under_soft_limit. +DOWNLOAD_SOFT_TIME_LIMIT = 1350 +DOWNLOAD_HARD_TIME_LIMIT = 1500 + + +def _finalize_soft_limited(session: SyncSession, source_id: int) -> None: + """Defense in depth for the soft-time-limit kill path. + + A SoftTimeLimitExceeded unwinds download_source before phase 3 can + finalize the DownloadEvent, leaving it 'running' until the recovery + sweep stamps a context-free "stranded" error 30 min later — AND + leaving backfill_runs_remaining undecremented so the source re-runs + and re-strands every tick (Anduo #39912, 2026-06-03). Flip the + in-flight event to error with a real reason, mirror phase 3's + source-health write, and decrement any backfill budget so a + chronically-slow source self-heals back to tick mode. + + The caller owns the commit. All mutations are gated on actually + finding a running event, so a benign late soft-limit (phase 3 already + committed) is a no-op. + """ + now = datetime.now(UTC) + ev = session.execute( + select(DownloadEvent) + .where(DownloadEvent.source_id == source_id) + .where(DownloadEvent.status == "running") + .order_by(DownloadEvent.id.desc()) + .limit(1) + ).scalar_one_or_none() + if ev is None: + return + ev.status = "error" + ev.finished_at = now + ev.error = ( + f"killed by Celery soft time limit ({DOWNLOAD_SOFT_TIME_LIMIT}s) " + "before the gallery-dl subprocess returned — the run exceeded its " + "budget and its stdout/stderr were lost with the worker thread. " + "If this recurs, the source is too large for one run; the backfill " + "budget was decremented so the next tick walks less." + ) + ev.metadata_ = { + **(ev.metadata_ or {}), + "error_type": "timeout", + "soft_time_limited": True, + } + src = session.get(Source, source_id) + if src is not None: + src.consecutive_failures = (src.consecutive_failures or 0) + 1 + src.last_error = "soft time limit exceeded" + src.error_type = "timeout" + src.last_checked_at = now + if (src.backfill_runs_remaining or 0) > 0: + src.backfill_runs_remaining = max(0, src.backfill_runs_remaining - 1) + @celery.task( name="backend.app.tasks.download.download_source", @@ -29,8 +104,8 @@ _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64" retry_backoff_max=120, retry_jitter=True, max_retries=3, - soft_time_limit=900, - time_limit=1200, + soft_time_limit=DOWNLOAD_SOFT_TIME_LIMIT, + time_limit=DOWNLOAD_HARD_TIME_LIMIT, ) def download_source(self, source_id: int) -> int: """Returns the DownloadEvent.id.""" @@ -73,4 +148,19 @@ def download_source(self, source_id: int) -> int: finally: await async_engine.dispose() - return asyncio.run(_run()) + try: + return asyncio.run(_run()) + except SoftTimeLimitExceeded: + # phase 3 never ran — salvage the in-flight event so the operator + # sees a real reason instead of the recovery sweep's generic + # "stranded" 30 min later (Anduo #39912). Best-effort: a failure + # here must not mask the timeout. Re-raise so Celery + the + # task_run signal handler still record the kill. + try: + SyncFactory = _sync_session_factory() + with SyncFactory() as session: + _finalize_soft_limited(session, source_id) + session.commit() + except Exception: # noqa: BLE001 — cleanup must not swallow the kill + log.exception("soft-limit finalize failed for source %s", source_id) + raise diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 0ae8342..ada9f3e 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -46,9 +46,12 @@ MAX_RECOVERY_ATTEMPTS = 3 ORPHAN_PENDING_THRESHOLD_MINUTES = 30 # DownloadEvent (pending|running) recovery threshold. download_source has -# time_limit=1200s (20 min); 30 min is 10 min past that, so a legitimately- -# running task is never killed by the sweep. Operator-confirmed 2026-05-29 -# after 43 sources stranded at "last check never" by the in-flight guard. +# 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 @@ -535,7 +538,7 @@ def recover_stalled_download_events() -> int: 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 1200s hard time_limit — the event stays in-flight forever. The next + 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. diff --git a/tests/test_download_source_task.py b/tests/test_download_source_task.py index 2a1a946..c63faee 100644 --- a/tests/test_download_source_task.py +++ b/tests/test_download_source_task.py @@ -1,9 +1,15 @@ -"""Smoke test that the FC-3c Celery task is registered and routed correctly. +"""Tests for the FC-3c download_source Celery task wrapper. -Mirrors test_tasks_register.py — Celery's `include=[...]` is lazy, so -the task module must be imported explicitly to trigger registration. +Covers registration/routing (smoke) plus the soft-time-limit salvage +path (audit 2026-06-03, Anduo #39912): a SoftTimeLimitExceeded must not +leave the DownloadEvent stranded empty for the recovery sweep. """ +from datetime import UTC, datetime + +import pytest +from sqlalchemy import select + # Side-effect import: the @celery.task decorator on download_source fires # at module import time and registers the task with the global instance. import backend.app.tasks.download # noqa: F401 @@ -18,3 +24,164 @@ def test_download_source_routes_to_download_queue(): routes = celery.conf.task_routes assert "backend.app.tasks.download.*" in routes assert routes["backend.app.tasks.download.*"]["queue"] == "download" + + +def test_timeout_ladder_keeps_subprocess_budgets_under_soft_limit(): + """Regression guard for Anduo #39912: every gallery-dl subprocess + budget MUST sit below download_source's Celery soft limit so + subprocess.run raises its own TimeoutExpired (which captures partial + logs + finalizes the event) BEFORE Celery's SoftTimeLimitExceeded + preempts it. soft must in turn sit below the hard SIGKILL cap.""" + from backend.app.services.gallery_dl import ( + BACKFILL_TIMEOUT_SECONDS, + _DEFAULT_GDL_TIMEOUT_SECONDS, + ) + from backend.app.tasks.download import ( + DOWNLOAD_HARD_TIME_LIMIT, + DOWNLOAD_SOFT_TIME_LIMIT, + ) + + assert _DEFAULT_GDL_TIMEOUT_SECONDS < DOWNLOAD_SOFT_TIME_LIMIT + assert BACKFILL_TIMEOUT_SECONDS < DOWNLOAD_SOFT_TIME_LIMIT + assert DOWNLOAD_SOFT_TIME_LIMIT < DOWNLOAD_HARD_TIME_LIMIT + + +def test_decorated_limits_match_module_constants(): + """The @celery.task decorator must use the audited constants, not + drifted literals.""" + from backend.app.tasks.download import ( + DOWNLOAD_HARD_TIME_LIMIT, + DOWNLOAD_SOFT_TIME_LIMIT, + download_source, + ) + + assert download_source.soft_time_limit == DOWNLOAD_SOFT_TIME_LIMIT + assert download_source.time_limit == DOWNLOAD_HARD_TIME_LIMIT + + +def _seed_running_event(db_sync, *, slug, backfill, failures=0): + from backend.app.models import Artist, DownloadEvent, Source + + artist = Artist(name=slug, slug=slug) + db_sync.add(artist) + db_sync.flush() + source = Source( + artist_id=artist.id, platform="patreon", + url=f"https://patreon.com/{slug}", enabled=True, + config_overrides={}, backfill_runs_remaining=backfill, + consecutive_failures=failures, + ) + db_sync.add(source) + db_sync.flush() + ev = DownloadEvent( + source_id=source.id, status="running", + started_at=datetime.now(UTC), + ) + db_sync.add(ev) + db_sync.flush() + return source, ev.id + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_finalize_soft_limited_flips_event_and_decrements_backfill(db_sync): + from backend.app.models import DownloadEvent, Source + from backend.app.tasks.download import _finalize_soft_limited + + source, event_id = _seed_running_event( + db_sync, slug="anduo", backfill=2, failures=0, + ) + + _finalize_soft_limited(db_sync, source.id) + + status, finished_at, error, meta = db_sync.execute( + select( + DownloadEvent.status, DownloadEvent.finished_at, + DownloadEvent.error, DownloadEvent.metadata_, + ).where(DownloadEvent.id == event_id) + ).one() + assert status == "error" + assert finished_at is not None + assert "soft time limit" in (error or "").lower() + assert meta.get("error_type") == "timeout" + assert meta.get("soft_time_limited") is True + + backfill, failures, error_type = db_sync.execute( + select( + Source.backfill_runs_remaining, Source.consecutive_failures, + Source.error_type, + ).where(Source.id == source.id) + ).one() + assert backfill == 1 # 2 -> 1, source self-heals toward tick mode + assert failures == 1 # 0 -> 1, mirrors phase-3 source-health write + assert error_type == "timeout" + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_finalize_soft_limited_is_noop_without_running_event(db_sync): + """A benign late soft-limit (phase 3 already committed → no running + event) must not touch source health or backfill budget.""" + from backend.app.models import DownloadEvent, Source + from backend.app.tasks.download import _finalize_soft_limited + + source, event_id = _seed_running_event( + db_sync, slug="noevent", backfill=3, failures=0, + ) + # Simulate phase 3 having already finalized the event. + ev = db_sync.get(DownloadEvent, event_id) + ev.status = "ok" + db_sync.flush() + + _finalize_soft_limited(db_sync, source.id) + + backfill, failures, error_type = db_sync.execute( + select( + Source.backfill_runs_remaining, Source.consecutive_failures, + Source.error_type, + ).where(Source.id == source.id) + ).one() + assert backfill == 3 # untouched + assert failures == 0 # untouched + assert error_type is None + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_download_source_catches_soft_limit_and_salvages_event( + db_sync, monkeypatch, +): + """End-to-end wiring: when the inner run raises SoftTimeLimitExceeded, + download_source's handler must flip the in-flight event to error + instead of letting it strand. Uses eager mode + a stubbed asyncio.run + so no real gallery-dl subprocess is spawned.""" + from celery.exceptions import SoftTimeLimitExceeded + + import backend.app.tasks.download as dl + from backend.app.models import DownloadEvent + + monkeypatch.setattr(celery.conf, "task_always_eager", True) + monkeypatch.setattr(celery.conf, "task_eager_propagates", False) + + source, event_id = _seed_running_event( + db_sync, slug="anduowire", backfill=1, failures=0, + ) + # The task opens a fresh session via _sync_session_factory(); commit + # so that session can see the seeded running event. + db_sync.commit() + + def _raise(coro=None, *a, **k): + # Close the un-awaited coroutine so pytest output stays pristine. + if coro is not None and hasattr(coro, "close"): + coro.close() + raise SoftTimeLimitExceeded("simulated soft limit") + + monkeypatch.setattr(dl.asyncio, "run", _raise) + + with pytest.raises(SoftTimeLimitExceeded): + dl.download_source.delay(source.id).get(propagate=True) + + status = db_sync.execute( + select(DownloadEvent.status).where(DownloadEvent.id == event_id) + ).scalar_one() + assert status == "error"