fix(download): escalating per-host cooldown on HTTP 429 (#1263)

Anna's Archive 429-throttles the source IP after repeated automated
requests.
The bypasser could clear the DDoS-Guard challenge but not the 429, so
each retry
re-solved, re-spawned Chrome, and rotated mirrors that share the same IP
- a
costly loop that never converged.

Add a process-global, per-host cooldown that escalates 2 -> 5 -> 10 ->
15 -> 30
minutes each time a host 429s again after its window elapsed, resetting
after a
long clear gap. Mirror selection skips cooling hosts and the bypasser
refuses to
solve one, so a throttled host fails fast instead of storming the
solver.

shelfmark/bypass/internal_bypasser.py

@@ -1400,6 +1400,11 @@ def _try_with_cached_cookies(url: str, hostname: str) -> str | None:
         if response.status_code == HTTPStatus.OK:
             logger.debug("Cached cookies worked, skipped Chrome bypass")
             return response.text
+        if response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
+            # Throttled, not challenged: arm the per-host backoff so the caller stops
+            # rotating into this host and re-solving. get_bypassed_page checks it before
+            # the next Chrome solve.
+            network.note_rate_limited(url)
         logger.debug(
             "Cached cookies rejected (%s) for %s; discarding them",
             response.status_code,
@@ -1438,6 +1443,18 @@ def get_bypassed_page(
     attempt_url = sel.rewrite(url)
     hostname = urlparse(attempt_url).hostname or ""
 
+    # A 429 means the origin is throttling this IP; the challenge still renders, so a
+    # solve "succeeds" but the cleared request is rejected again and the throttle is only
+    # renewed. Never spend a minutes-long Chrome solve on a cooling-down host - fail fast
+    # so the caller waits the backoff out instead of looping the solve.
+    remaining = network.host_cooldown_remaining(attempt_url)
+    if remaining > 0:
+        msg = (
+            f"{hostname} is rate-limited (429); skipping bypass for ~{remaining:.0f}s "
+            "until the cooldown clears."
+        )
+        raise network.RateLimitedError(msg)
+
     cached_result = _try_with_cached_cookies(attempt_url, hostname)
     if cached_result:
         return cached_result

shelfmark/download/http.py

@@ -52,6 +52,7 @@ _BYPASSER_ERRORS = (
     RuntimeError,
     TypeError,
     ValueError,
+    network.RateLimitedError,
     requests.exceptions.RequestException,
 )
 
@@ -377,6 +378,17 @@ def html_get_page(
                 "not solved. Check that FlareSolverr/the CF bypasser is reachable.",
                 bypass_url,
             )
+        except network.RateLimitedError as e:
+            # Not a bypasser malfunction: the host is throttling this IP and a solve
+            # cannot help. Surface the wait as a plain failure so the search ends cleanly
+            # instead of looping another minutes-long solve against a 429.
+            logger.info("Skipping bypass (rate-limited): %s", e)
+            if status_callback:
+                try:
+                    status_callback("resolving", "Rate limited, try again shortly")
+                except _STATUS_CALLBACK_ERRORS:
+                    logger.debug("Rate-limit status callback failed", exc_info=True)
+            return _fail(str(e), bypass_url)
         except _BYPASSER_ERRORS as e:
             logger.warning("Bypasser error: %s: %s", type(e).__name__, e)
             # Surface the real reason. Without this the caller only sees an empty
@@ -698,6 +710,12 @@ def html_get_page(
                     f"Anna's Archive returned 404 Not Found for {current_url}.", current_url
                 )
 
+            # 429 = origin throttling this IP. Arm the per-host backoff so selection and
+            # the bypasser stop hammering it, then fall through to normal rotation onto a
+            # mirror that is not (yet) rate-limited.
+            if status == _HTTP_STATUS_RATE_LIMITED:
+                network.note_rate_limited(current_url)
+
             # Try mirror/DNS rotation on retryable errors. A failure that proves the
             # mirror is unusable also drops it from this process's rotation, so the
             # next search does not pay for it again.
@@ -851,6 +869,7 @@ def download_url(
             # Rate limited - skip to next source immediately
             # (waiting doesn't help with concurrent downloads hitting the same server)
             if status == _HTTP_STATUS_RATE_LIMITED:
+                network.note_rate_limited(current_url)
                 logger.info("Rate limited (429) - trying next source")
                 if status_callback:
                     status_callback("resolving", "Server busy, trying next")

shelfmark/download/network.py

@@ -3,12 +3,13 @@
 import fnmatch
 import ipaddress
 import socket
+import time
 import urllib.parse
 import urllib.request
 from datetime import UTC, datetime, timedelta
 from http import HTTPStatus
 from socket import AddressFamily, SocketKind
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, NamedTuple, cast
 
 import dns.resolver
 import httpx
@@ -287,6 +288,108 @@ _dead_aa_urls: set[str] = set()
 _dead_aa_urls_lock = _RLock()
 
 
+# Per-host rate-limit backoff. A 429 is the origin throttling *this IP*, not a challenge:
+# a DDoS-Guard/Cloudflare solve still renders, so the bypass "succeeds" yet the cleared
+# request is rejected again and the throttle is only renewed. The single answer is to
+# wait, so a 429 sidelines the host for a growing window - mirror selection and the
+# bypasser both skip a cooling-down host until its deadline passes. The wait escalates
+# 2 -> 5 -> 10 -> 15 -> 30 minutes each time the host throttles us again *after* we
+# already waited a full window out; a host left clear for longer than the top step
+# starts the ladder over. Keyed by host so every mirror and source shares one view;
+# in-memory only, so a restart starts clean.
+_RATE_LIMIT_COOLDOWN_LADDER_SECONDS: tuple[float, ...] = (120.0, 300.0, 600.0, 900.0, 1800.0)
+# A host that has been clear this long is treated as a fresh episode: the next 429
+# restarts the ladder at 2 minutes rather than resuming the escalation.
+_RATE_LIMIT_RESET_AFTER_SECONDS = 1800.0
+
+
+class _Cooldown(NamedTuple):
+    """One host's active rate-limit window and how far up the ladder it has climbed."""
+
+    deadline: float  # time.monotonic() value at which the wait expires
+    level: int  # index into _RATE_LIMIT_COOLDOWN_LADDER_SECONDS
+
+
+_host_cooldowns: dict[str, _Cooldown] = {}
+_host_cooldowns_lock = _RLock()
+
+
+class RateLimitedError(Exception):
+    """Raised to abandon a request whose host is in a 429 cooldown.
+
+    Not a transport failure - nothing is wrong with the network, the origin is
+    throttling this IP and only time clears it. Callers surface it as a plain failure
+    rather than retrying or handing the URL to the bypasser.
+    """
+
+
+def _cooldown_key(url: str) -> str:
+    """Host a cooldown is keyed by; '' when the URL carries none."""
+    return (urllib.parse.urlparse(url).hostname or "").lower()
+
+
+def note_rate_limited(url: str) -> float:
+    """Escalate a host's 429 backoff and (re)arm its cooldown; return the wait applied.
+
+    The step advances only when a fresh 429 arrives *after* the previous window already
+    elapsed - i.e. we waited it out and the host throttled us again. A 429 that lands
+    while the host is still cooling is the same episode: it neither escalates the level
+    nor shortens the wait. See the ladder note above.
+    """
+    host = _cooldown_key(url)
+    if not host:
+        return 0.0
+    now = time.monotonic()
+    ladder = _RATE_LIMIT_COOLDOWN_LADDER_SECONDS
+    with _host_cooldowns_lock:
+        prev = _host_cooldowns.get(host)
+        if prev is not None and now < prev.deadline:
+            # Still inside the current window - same throttling episode, leave it be.
+            return prev.deadline - now
+        if prev is None or now - prev.deadline > _RATE_LIMIT_RESET_AFTER_SECONDS:
+            level = 0
+        else:
+            level = min(prev.level + 1, len(ladder) - 1)
+        wait = ladder[level]
+        _host_cooldowns[host] = _Cooldown(deadline=now + wait, level=level)
+    logger.info(
+        "Rate limited (429): backing off %s for %.0fs (step %d/%d)",
+        host,
+        wait,
+        level + 1,
+        len(ladder),
+    )
+    return wait
+
+
+def host_cooldown_remaining(url: str) -> float:
+    """Seconds left on a host's 429 cooldown; 0.0 when clear or expired.
+
+    Leaves an expired record in place: the ladder level it carries is what a later 429
+    escalates from (or resets, once the clear gap is long enough).
+    """
+    host = _cooldown_key(url)
+    if not host:
+        return 0.0
+    now = time.monotonic()
+    with _host_cooldowns_lock:
+        rec = _host_cooldowns.get(host)
+        if rec is None or rec.deadline <= now:
+            return 0.0
+        return rec.deadline - now
+
+
+def is_host_cooling_down(url: str) -> bool:
+    """True while ``url``'s host is inside its 429 cooldown window."""
+    return host_cooldown_remaining(url) > 0.0
+
+
+def clear_host_cooldowns() -> None:
+    """Forget all rate-limit cooldowns (manual reset / tests)."""
+    with _host_cooldowns_lock:
+        _host_cooldowns.clear()
+
+
 def _ensure_initialized() -> None:
     """Lazy guard so runtime setup happens once and late calls still work."""
     global _initialized
@@ -1418,8 +1521,13 @@ def get_available_aa_urls() -> list[str]:
         if not alive and _aa_urls:
             logger.warning("All AA mirrors quarantined; retrying the full list")
             _dead_aa_urls.clear()
-            return _aa_urls.copy()
-    return alive
+            alive = _aa_urls.copy()
+    # Prefer mirrors that are not serving a 429 cooldown so rotation stops hammering a
+    # throttled host. When every live mirror is cooling, keep the full live list rather
+    # than returning nothing: selection must never be left with nowhere to point, and
+    # the bypasser's fail-fast reports the "all rate-limited" case with a clear error.
+    breathing = [url for url in alive if not is_host_cooling_down(url)]
+    return breathing or alive
 
 
 def _aa_base_for_url(url: str) -> str:

tests/download/test_network_rate_limit_cooldown.py

@@ -0,0 +1,123 @@
+"""Tests for the per-host 429 backoff.
+
+A 429 is the origin throttling this IP, which a challenge solve cannot clear - so the
+host is sidelined for a growing window (2 -> 5 -> 10 -> 15 -> 30 min) that escalates
+only when the host throttles us again *after* a full window has already elapsed. Mirror
+selection skips a cooling-down host; the bypasser refuses to solve one. These guard that
+the ladder climbs, caps, resets after a long clear gap, and stays per host.
+"""
+
+import shelfmark.download.network as network
+
+MIRRORS = ["https://aa-one.test", "https://aa-two.test", "https://aa-three.test"]
+LADDER = (120.0, 300.0, 600.0, 900.0, 1800.0)
+
+
+def _fresh(monkeypatch, *, urls=None, start=1000.0):
+    """Reset cooldown state and install a controllable monotonic clock.
+
+    Returns a ``clock`` list whose single element is the current fake time; mutate
+    ``clock[0]`` to advance it.
+    """
+    clock = [start]
+    monkeypatch.setattr(network, "_host_cooldowns", {})
+    monkeypatch.setattr(network.time, "monotonic", lambda: clock[0])
+    if urls is not None:
+        monkeypatch.setattr(network, "_initialized", True)
+        monkeypatch.setattr(network, "_aa_urls", list(urls))
+        monkeypatch.setattr(network, "_aa_base_url", urls[0])
+        monkeypatch.setattr(network, "_current_aa_url_index", 0)
+        monkeypatch.setattr(network, "_dead_aa_urls", set())
+    return clock
+
+
+def test_first_429_arms_the_two_minute_step(monkeypatch):
+    _fresh(monkeypatch)
+
+    assert network.note_rate_limited("https://h.test/search?q=dune") == 120.0
+    # Keyed by host: any URL on the same host reads the same cooldown.
+    assert network.is_host_cooling_down("https://h.test/other") is True
+    assert network.host_cooldown_remaining("https://h.test") == 120.0
+
+
+def test_cooldown_expires_after_the_window(monkeypatch):
+    clock = _fresh(monkeypatch)
+
+    network.note_rate_limited("https://h.test")
+    clock[0] += 121
+
+    assert network.is_host_cooling_down("https://h.test") is False
+    assert network.host_cooldown_remaining("https://h.test") == 0.0
+
+
+def test_re_offense_after_expiry_climbs_the_ladder(monkeypatch):
+    clock = _fresh(monkeypatch)
+
+    for expected in LADDER:
+        assert network.note_rate_limited("https://h.test") == expected
+        # Wait the whole window out, then get throttled again -> next step.
+        clock[0] += expected + 1
+
+    # Top step holds: further re-offenses stay at 30 minutes, never beyond.
+    assert network.note_rate_limited("https://h.test") == 1800.0
+
+
+def test_429_while_still_cooling_does_not_escalate(monkeypatch):
+    clock = _fresh(monkeypatch)
+
+    assert network.note_rate_limited("https://h.test") == 120.0
+    clock[0] += 30  # still inside the first window
+
+    # Same episode: keep the remaining wait, do not advance the ladder.
+    assert network.note_rate_limited("https://h.test") == 90.0
+    clock[0] += 91  # let the (unchanged) 2-min window lapse
+    # The next post-expiry 429 is step 2, proving the mid-window hit did not escalate.
+    assert network.note_rate_limited("https://h.test") == 300.0
+
+
+def test_long_clear_gap_restarts_the_ladder(monkeypatch):
+    clock = _fresh(monkeypatch)
+
+    network.note_rate_limited("https://h.test")  # step 1: 120s
+    clock[0] += 120 + 1801  # window lapses, then a gap longer than the reset threshold
+
+    assert network.note_rate_limited("https://h.test") == 120.0
+
+
+def test_backoff_is_per_host(monkeypatch):
+    _fresh(monkeypatch)
+
+    network.note_rate_limited("https://a.test")
+    assert network.is_host_cooling_down("https://a.test") is True
+    assert network.is_host_cooling_down("https://b.test") is False
+
+
+def test_available_mirrors_skip_cooling_hosts(monkeypatch):
+    _fresh(monkeypatch, urls=MIRRORS)
+
+    network.note_rate_limited(MIRRORS[1])
+    assert network.get_available_aa_urls() == [MIRRORS[0], MIRRORS[2]]
+
+
+def test_all_mirrors_cooling_falls_back_to_full_list(monkeypatch):
+    _fresh(monkeypatch, urls=MIRRORS)
+
+    for mirror in MIRRORS:
+        network.note_rate_limited(mirror)
+    # Never leave selection with nowhere to point; the bypasser fail-fast handles this.
+    assert network.get_available_aa_urls() == MIRRORS
+
+
+def test_clear_host_cooldowns_resets_everything(monkeypatch):
+    _fresh(monkeypatch)
+
+    network.note_rate_limited("https://h.test")
+    network.clear_host_cooldowns()
+    assert network.is_host_cooling_down("https://h.test") is False
+
+
+def test_urls_without_a_host_are_ignored(monkeypatch):
+    _fresh(monkeypatch)
+
+    assert network.note_rate_limited("not-a-url") == 0.0
+    assert network.is_host_cooling_down("not-a-url") is False