From 87c7318125f12c587b16714ead802712792c9994 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 21:22:32 -0400 Subject: [PATCH 1/2] 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() -- 2.52.0 From a3c9499e931101a9b7f5a499593b4c2ad8bcdf2b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 21:22:32 -0400 Subject: [PATCH 2/2] feat(subs): kick off the first backfill walk immediately on source create A new enabled source is armed for run-until-done backfill (#693) but would sit idle until the next scheduler tick (~60s). create_source now enqueues the first walk right away (pending DownloadEvent + download_source.delay), skipping only when the platform is in a rate-limit cooldown (the scheduler picks it up when that clears). Disabled sources still don't dispatch. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/api/sources.py | 16 ++++++++++++++++ tests/test_api_sources.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/backend/app/api/sources.py b/backend/app/api/sources.py index d5871e1..55c8a3d 100644 --- a/backend/app/api/sources.py +++ b/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 diff --git a/tests/test_api_sources.py b/tests/test_api_sources.py index 71f4dde..e136ac2 100644 --- a/tests/test_api_sources.py +++ b/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={ -- 2.52.0