Merge pull request 'Patreon download concurrency cap + immediate backfill kickoff on create' (#76) from dev into main

backend/app/api/sources.py

@@ -85,6 +85,22 @@ async def create_source():
             return _bad("empty_url", detail=str(exc))
         except DuplicateSourceError as exc:
             return _bad("duplicate", status=409, existing_id=exc.existing_id)
+
+        # Immediate kickoff: a new enabled source is armed for backfill (#693)
+        # but would otherwise sit idle until the next scheduler tick (~60s).
+        # Enqueue the first walk now, skipping only if the platform is in a
+        # rate-limit cooldown (the scheduler picks it up when that clears).
+        dispatch_id = None
+        if record.enabled:
+            cooldowns = await active_platform_cooldowns(session)
+            if record.platform not in cooldowns:
+                session.add(DownloadEvent(source_id=record.id, status="pending"))
+                await session.commit()
+                dispatch_id = record.id
+
+    if dispatch_id is not None:
+        from ..tasks.download import download_source
+        download_source.delay(dispatch_id)
     return jsonify(record.to_dict()), 201
 
 

backend/app/services/platform_lock.py

@@ -0,0 +1,59 @@
+"""Per-platform download concurrency cap.
+
+Some platforms (Patreon) are API-rate-sensitive enough that two *simultaneous*
+walks can trip the server's rate limit even with each source pacing its own
+requests. The platform-cooldown handles the AFTERMATH of a 429; this is the
+preventive half — it serializes downloads PER PLATFORM to one at a time.
+
+Different platforms still run concurrently up to the worker's concurrency; only
+a second walk on the SAME serialized platform waits. The lock lives in Redis
+(the Celery broker) with a TTL, so a SIGKILL'd worker can't wedge a platform —
+the lock auto-expires shortly after the download hard time limit.
+"""
+from __future__ import annotations
+
+import logging
+
+import redis
+
+from ..config import get_config
+
+log = logging.getLogger(__name__)
+
+# Platforms walked one-at-a-time. gallery-dl platforms are intentionally NOT
+# here: each runs as a self-pacing subprocess and they're lower-volume. Add a
+# platform here to cap it to a single concurrent walk.
+SERIALIZED_PLATFORMS = frozenset({"patreon"})
+
+_LOCK_PREFIX = "fc:download_lock:"
+
+_client: redis.Redis | None = None
+
+
+def _redis() -> redis.Redis:
+    # One client per worker process (Celery prefork forks before tasks run, so
+    # each process lazily builds its own). redis-py pools connections.
+    global _client
+    if _client is None:
+        _client = redis.from_url(get_config().celery_broker_url)
+    return _client
+
+
+def platform_lock(platform: str, *, ttl_seconds: int):
+    """A non-blocking Redis lock for `platform`, or None when the platform is
+    not serialized. Caller does `.acquire(blocking=False)` / `.release()`.
+
+    Returns None (rather than raising) on any Redis error so a broker hiccup
+    degrades to the prior behaviour (uncapped) instead of stalling downloads.
+    """
+    if platform not in SERIALIZED_PLATFORMS:
+        return None
+    try:
+        return _redis().lock(
+            f"{_LOCK_PREFIX}{platform}",
+            timeout=ttl_seconds,
+            blocking=False,
+        )
+    except redis.RedisError as exc:  # pragma: no cover - broker outage
+        log.warning("platform_lock unavailable for %s: %s", platform, exc)
+        return None

backend/app/tasks/download.py

@@ -44,6 +44,25 @@ _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64"
 DOWNLOAD_SOFT_TIME_LIMIT = 1350
 DOWNLOAD_HARD_TIME_LIMIT = 1500
 
+# Per-platform serialization (plan: concurrency cap). When a serialized
+# platform (Patreon) is already walking, defer this run by re-enqueuing it a
+# little later rather than holding a worker slot or bowling into the same rate
+# limit. Bounded so a wedged platform eventually runs anyway. The lock TTL sits
+# just past the hard kill so a SIGKILL'd worker's lock auto-expires; a backfill
+# chunk only holds it ~10 min, so the bounded wait stays well under the 30-min
+# DownloadEvent recovery sweep.
+_PLATFORM_LOCK_TTL = DOWNLOAD_HARD_TIME_LIMIT + 120
+_SERIALIZE_COUNTDOWN = 20
+_MAX_SERIALIZE_WAITS = 45  # ~15 min ceiling, then run uncapped as a safety valve
+
+
+def _peek_platform(source_id: int) -> str | None:
+    SyncFactory = _sync_session_factory()
+    with SyncFactory() as session:
+        return session.execute(
+            select(Source.platform).where(Source.id == source_id)
+        ).scalar_one_or_none()
+
 
 def _finalize_soft_limited(session: SyncSession, source_id: int) -> None:
     """Defense in depth for the soft-time-limit kill path.
@@ -107,9 +126,48 @@ def _finalize_soft_limited(session: SyncSession, source_id: int) -> None:
     soft_time_limit=DOWNLOAD_SOFT_TIME_LIMIT,
     time_limit=DOWNLOAD_HARD_TIME_LIMIT,
 )
-def download_source(self, source_id: int) -> int:
+def download_source(self, source_id: int, _serialize_waits: int = 0) -> int:
     """Returns the DownloadEvent.id."""
 
+    # Per-platform concurrency cap: only one Patreon walk runs at a time.
+    from ..services.platform_lock import platform_lock
+
+    platform = _peek_platform(source_id)
+    lock = (
+        platform_lock(platform, ttl_seconds=_PLATFORM_LOCK_TTL)
+        if platform is not None
+        else None
+    )
+    if lock is not None:
+        try:
+            got_lock = bool(lock.acquire(blocking=False))
+        except Exception:  # noqa: BLE001 — broker hiccup → degrade to uncapped
+            log.warning("platform lock acquire failed for %s; running uncapped", platform)
+            lock = None
+            got_lock = True
+        if lock is not None and not got_lock:
+            if _serialize_waits < _MAX_SERIALIZE_WAITS:
+                # Another walk on this platform holds the lock — re-enqueue
+                # ourselves shortly (the pending DownloadEvent stays; no new
+                # event, no log spam) rather than running concurrently into
+                # the rate limit.
+                download_source.apply_async(
+                    (source_id,),
+                    {"_serialize_waits": _serialize_waits + 1},
+                    countdown=_SERIALIZE_COUNTDOWN,
+                )
+                log.info(
+                    "download_source(%s) deferred — %s already walking (wait %d/%d)",
+                    source_id, platform, _serialize_waits + 1, _MAX_SERIALIZE_WAITS,
+                )
+                return -1
+            log.warning(
+                "download_source(%s) running WITHOUT the %s lock — max serialize "
+                "waits hit, proceeding uncapped",
+                source_id, platform,
+            )
+            lock = None
+
     async def _run():
         async_factory, async_engine = async_session_factory()
         SyncFactory = _sync_session_factory()
@@ -169,3 +227,12 @@ def download_source(self, source_id: int) -> int:
         except Exception:  # noqa: BLE001 — cleanup must not swallow the kill
             log.exception("soft-limit finalize failed for source %s", source_id)
         raise
+    finally:
+        # Release the per-platform lock so the next walk can proceed. The TTL is
+        # a backstop for a SIGKILL; here we free it the instant the run ends. A
+        # late release after TTL expiry raises (lock already gone) — ignore it.
+        if lock is not None:
+            try:
+                lock.release()
+            except Exception:  # noqa: BLE001 — TTL may have already freed it
+                pass

tests/test_api_sources.py

@@ -89,6 +89,39 @@ async def test_create_list_get_delete(client, artist):
     assert (await client.get(f"/api/sources/{sid}")).status_code == 404
 
 
+@pytest.mark.asyncio
+async def test_create_kicks_off_backfill_for_enabled_source(client, artist, monkeypatch):
+    """A new enabled source dispatches its first walk immediately (no waiting
+    for the next scheduler tick), unless its platform is in cooldown."""
+    delays: list = []
+    monkeypatch.setattr(
+        "backend.app.tasks.download.download_source.delay",
+        lambda *a, **k: delays.append(a),
+    )
+    resp = await client.post("/api/sources", json={
+        "artist_id": artist.id, "platform": "patreon",
+        "url": "https://patreon.com/kickoff",
+    })
+    assert resp.status_code == 201
+    sid = (await resp.get_json())["id"]
+    assert delays == [(sid,)]
+
+
+@pytest.mark.asyncio
+async def test_create_disabled_source_does_not_kick_off(client, artist, monkeypatch):
+    delays: list = []
+    monkeypatch.setattr(
+        "backend.app.tasks.download.download_source.delay",
+        lambda *a, **k: delays.append(a),
+    )
+    resp = await client.post("/api/sources", json={
+        "artist_id": artist.id, "platform": "patreon",
+        "url": "https://patreon.com/disabled", "enabled": False,
+    })
+    assert resp.status_code == 201
+    assert delays == []
+
+
 @pytest.mark.asyncio
 async def test_create_rejects_unknown_platform(client, artist):
     resp = await client.post("/api/sources", json={

tests/test_download_source_task.py

@@ -162,6 +162,12 @@ async def test_download_source_catches_soft_limit_and_salvages_event(
 
     monkeypatch.setattr(celery.conf, "task_always_eager", True)
     monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+    # This test exercises the soft-limit salvage path, not the per-platform
+    # serialization. Neutralize the Redis lock so it can't defer/recurse under
+    # eager mode (and so it doesn't contend with test_platform_lock).
+    monkeypatch.setattr(
+        "backend.app.services.platform_lock.platform_lock", lambda *a, **k: None,
+    )
 
     source, event_id = _seed_running_event(
         db_sync, slug="anduowire", backfill=1, failures=0,

tests/test_platform_lock.py

@@ -0,0 +1,30 @@
+"""Per-platform download concurrency cap (Redis lock)."""
+import pytest
+
+from backend.app.services.platform_lock import platform_lock
+
+# Needs the broker (Redis) the lock lives in.
+pytestmark = pytest.mark.integration
+
+
+def test_non_serialized_platform_has_no_lock():
+    # gallery-dl platforms aren't capped — they get no lock at all.
+    assert platform_lock("subscribestar", ttl_seconds=60) is None
+    assert platform_lock("deviantart", ttl_seconds=60) is None
+
+
+def test_patreon_serializes_to_one_holder():
+    first = platform_lock("patreon", ttl_seconds=60)
+    assert first is not None
+    assert first.acquire(blocking=False) is True
+    try:
+        # A second acquirer is refused while the first holds it.
+        second = platform_lock("patreon", ttl_seconds=60)
+        assert second.acquire(blocking=False) is False
+    finally:
+        first.release()
+
+    # Once released, the platform is acquirable again.
+    third = platform_lock("patreon", ttl_seconds=60)
+    assert third.acquire(blocking=False) is True
+    third.release()