"""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