From 87c7318125f12c587b16714ead802712792c9994 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 21:22:32 -0400 Subject: [PATCH] feat(downloads): serialize same-platform downloads (Patreon concurrency cap) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two concurrent Patreon walks could trip the server rate limit even with each source pacing its own requests. The platform-cooldown handled the aftermath of a 429; this adds the preventive half — a per-platform Redis lock so only one Patreon walk runs at a time. Different platforms still run concurrently up to the worker concurrency; only a second walk on the SAME serialized platform waits. download_source acquires fc:download_lock: (non-blocking) before the run. On contention it re-enqueues itself with a short countdown (the pending event stays — no new event, no log spam), bounded to ~15 min then runs uncapped as a safety valve. The lock TTL sits just past the hard kill so a SIGKILL'd worker auto-releases; a backfill chunk only holds it ~10 min, well under the 30-min DownloadEvent recovery sweep. A broker hiccup degrades to uncapped (prior behaviour) rather than stalling downloads. SERIALIZED_PLATFORMS={patreon}; gallery-dl platforms are left uncapped (self-pacing subprocesses). Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/platform_lock.py | 59 +++++++++++++++++++++++ backend/app/tasks/download.py | 69 ++++++++++++++++++++++++++- tests/test_download_source_task.py | 6 +++ tests/test_platform_lock.py | 30 ++++++++++++ 4 files changed, 163 insertions(+), 1 deletion(-) create mode 100644 backend/app/services/platform_lock.py create mode 100644 tests/test_platform_lock.py diff --git a/backend/app/services/platform_lock.py b/backend/app/services/platform_lock.py new file mode 100644 index 0000000..c70e1e8 --- /dev/null +++ b/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 diff --git a/backend/app/tasks/download.py b/backend/app/tasks/download.py index 42c075d..17d04e3 100644 --- a/backend/app/tasks/download.py +++ b/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 diff --git a/tests/test_download_source_task.py b/tests/test_download_source_task.py index 7c709ca..22d59dd 100644 --- a/tests/test_download_source_task.py +++ b/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, diff --git a/tests/test_platform_lock.py b/tests/test_platform_lock.py new file mode 100644 index 0000000..161145a --- /dev/null +++ b/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()