Merge pull request #3482 from DevYukine/feat/playmatch-rate-limit

feat(playmatch): add ratelimiting

Author: gantoine

backend/handler/metadata/playmatch_handler.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 from enum import Enum
 from typing import Final, NotRequired, TypedDict
@@ -12,6 +13,12 @@
 from models.rom import Rom, RomFile
 from utils import get_version
 from utils.context import ctx_httpx_client
+from utils.rate_limiter import RateLimiter
+
+# Playmatch caps clients at 4 req/s per IP
+PLAYMATCH_MAX_REQUESTS_PER_SECOND: Final[float] = 4
+PLAYMATCH_MAX_REQUEST_ATTEMPTS: Final[int] = 2
+_rate_limiter = RateLimiter(PLAYMATCH_MAX_REQUESTS_PER_SECOND)
 
 
 class PlaymatchProvider(str, Enum):
@@ -144,29 +151,47 @@ async def _request(self, url: str, query: dict) -> dict:
 
         headers = {"user-agent": f"RomM/{get_version()}"}
 
-        try:
-            res = await httpx_client.get(
-                str(url_with_query), headers=headers, timeout=60
-            )
-            res.raise_for_status()
-            return res.json()
-        except (httpx.HTTPStatusError, httpx.ConnectError, httpx.ReadTimeout) as exc:
-            log.warning("Connection error: can't connect to Playmatch", exc_info=True)
-            raise HTTPException(
-                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-                detail="Can't connect to Playmatch, check your internet connection",
-            ) from exc
-        except json.JSONDecodeError as exc:
-            log.error("Error decoding JSON response from Playmatch: %s", exc)
-            return {}
+        for attempt in range(PLAYMATCH_MAX_REQUEST_ATTEMPTS):
+            await _rate_limiter.acquire()
+            try:
+                res = await httpx_client.get(
+                    str(url_with_query), headers=headers, timeout=60
+                )
+                res.raise_for_status()
+                return res.json()
+            except (
+                httpx.HTTPStatusError,
+                httpx.ConnectError,
+                httpx.ReadTimeout,
+            ) as exc:
+                if (
+                    attempt == 0
+                    and isinstance(exc, httpx.HTTPStatusError)
+                    and exc.response.status_code == status.HTTP_429_TOO_MANY_REQUESTS
+                ):
+                    log.warning("Playmatch: rate limit hit, retrying after 2s")
+                    await asyncio.sleep(2)
+                    continue
+                log.warning(
+                    "Connection error: can't connect to Playmatch", exc_info=True
+                )
+                raise HTTPException(
+                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                    detail="Can't connect to Playmatch, check your internet connection",
+                ) from exc
+            except json.JSONDecodeError as exc:
+                log.error("Error decoding JSON response from Playmatch: %s", exc)
+                return {}
+
+        return {}
 
     async def lookup_rom(self, files: list[RomFile]) -> PlaymatchRomMatch:
         """
         Identify a ROM file using Playmatch API.
 
         :param rom_attrs: A dictionary containing the ROM attributes.
-        :return: A PlaymatchRomMatch objects containing the matched ROM information.
-        :raises HTTPException: If the request fails or the service is unavailable.
+        :return: A PlaymatchRomMatch with the matched IDs, or an empty match if the
+            lookup fails. Playmatch is best-effort and never raises to the caller.
         """
         fallback_rom = PlaymatchRomMatch(
             igdb_id=None,
@@ -203,8 +228,9 @@ async def lookup_rom(self, files: list[RomFile]) -> PlaymatchRomMatch:
                     "sha1": first_file.sha1_hash,
                 },
             )
-        except httpx.HTTPStatusError:
+        except Exception as exc:
             # We silently fail if the service is unavailable as this should not block the rest of RomM.
+            log.warning("Playmatch lookup failed, skipping: %s", exc)
             return fallback_rom
 
         game_match_type = response.get("gameMatchType", None)
@@ -290,6 +316,7 @@ async def submit_manual_match_suggestion(self, rom: Rom) -> None:
             }
 
             httpx_client = ctx_httpx_client.get()
+            await _rate_limiter.acquire()
             res = await httpx_client.post(
                 self.suggestion_url,
                 json=payload,

backend/tests/utils/test_rate_limiter.py

@@ -0,0 +1,57 @@
+import asyncio
+
+import pytest
+
+from utils.rate_limiter import RateLimiter
+
+
+def _record_sleeps(monkeypatch) -> list[float]:
+    """Replace asyncio.sleep with a no-op that records requested delays."""
+    sleeps: list[float] = []
+
+    async def fake_sleep(delay: float) -> None:
+        sleeps.append(delay)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
+    return sleeps
+
+
+class TestRateLimiter:
+    async def test_first_acquire_does_not_sleep(self, monkeypatch):
+        sleeps = _record_sleeps(monkeypatch)
+        limiter = RateLimiter(requests_per_second=4)
+
+        await limiter.acquire()
+
+        assert sleeps == []
+
+    async def test_slots_advance_by_interval(self, monkeypatch):
+        sleeps = _record_sleeps(monkeypatch)
+        limiter = RateLimiter(requests_per_second=4)
+
+        for _ in range(5):
+            await limiter.acquire()
+
+        # First grant is immediate; the next four are spaced one interval apart.
+        assert len(sleeps) == 4
+        for index, delay in enumerate(sleeps, start=1):
+            assert delay == pytest.approx(index * 0.25, abs=0.05)
+
+    async def test_concurrent_callers_are_spaced(self):
+        rate = 50
+        interval = 1 / rate
+        count = 5
+        limiter = RateLimiter(requests_per_second=rate)
+
+        loop = asyncio.get_running_loop()
+        start = loop.time()
+        await asyncio.gather(*(limiter.acquire() for _ in range(count)))
+        elapsed = loop.time() - start
+
+        assert elapsed >= (count - 1) * interval
+        assert elapsed < (count - 1) * interval + 0.5
+
+    @pytest.mark.parametrize("rate", [0, -1, -0.5])
+    def test_non_positive_rate_raises(self, rate):
+        with pytest.raises(ValueError):
+            RateLimiter(requests_per_second=rate)

backend/utils/rate_limiter.py

@@ -0,0 +1,28 @@
+import asyncio
+
+
+class RateLimiter:
+    """Pre-emptive async rate limiter.
+
+    Spaces grants so callers stay at or below ``requests_per_second``. Concurrent
+    callers each reserve the next slot before awaiting, so they are evenly spaced
+    instead of being released in a burst.
+    """
+
+    def __init__(self, requests_per_second: float) -> None:
+        if requests_per_second <= 0:
+            raise ValueError("requests_per_second must be positive")
+        self._min_interval = 1.0 / requests_per_second
+        self._next_slot = 0.0
+
+    async def acquire(self) -> None:
+        # Reserving the slot is a read-then-write on _next_slot with no await
+        # between, so it is atomic on the single-threaded loop and needs no lock.
+        loop = asyncio.get_running_loop()
+        now = loop.time()
+        slot = max(now, self._next_slot)
+        self._next_slot = slot + self._min_interval
+
+        delay = slot - now
+        if delay > 0:
+            await asyncio.sleep(delay)