fix(download): salvage soft-time-limit kills + fix timeout ladder

Backfill downloads stranded with empty logs + a generic "stranded by
recovery sweep" error. Root cause: the backfill gallery-dl subprocess
timeout (1170s) exceeded download_source's Celery soft_time_limit (900s),
so SoftTimeLimitExceeded preempted subprocess.TimeoutExpired. The
TimeoutExpired path (which captures partial stdout/stderr and finalizes
the event) never ran, the event was left 'running', and phase 3 never
decremented backfill_runs_remaining — so the source re-ran and
re-stranded every tick (Anduo #39912).

Two layers:
1. Raise download_source limits (soft 900→1350, hard 1200→1500) so both
   subprocess budgets (870 tick / 1170 backfill) sit below the soft
   limit with phase-3 persist headroom. Promote to module constants and
   guard the invariant with a test.
2. Catch SoftTimeLimitExceeded in download_source and finalize the
   in-flight event with a real reason, mirror phase-3 source-health, and
   decrement backfill so a chronically-slow source self-heals to tick
   mode. The existing celery_signals handler only covered TaskRun, not
   DownloadEvent — that was the gap.

Updates stale 900/1200 references in gallery_dl.py + maintenance.py.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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"