diff --git a/backend/app/routes/search.py b/backend/app/routes/search.py index a95c7850..bc1bf444 100644 --- a/backend/app/routes/search.py +++ b/backend/app/routes/search.py @@ -65,7 +65,10 @@ _refunds = RateLimiter(max_events=REFUNDS_PER_HOUR, window_seconds=3600) # The API source's budget check and the search.list call it pays for run one at a time: checked # apart, searches sent together at the last 100 units all passed, and all spent. One process # serves the API (a single uvicorn worker, as `ratelimit` says), so a process lock holds them all. +# A search waits this long for it, then is refused as busy: each waiter holds a request worker, and +# a slow YouTube (30 s a call at worst) would otherwise park them all behind one another. _api_search = threading.Lock() +API_SEARCH_WAIT_SECONDS = 10.0 def _gather( @@ -137,6 +140,14 @@ def _budget_spent() -> None: ) +def _busy() -> None: + """Say that a live search was refused because other API searches held the line too long.""" + log.info( + "a live search was refused: other searches held the line", + extra=logs.event("search.refused", reason="busy", waited=API_SEARCH_WAIT_SECONDS), + ) + + def _cap_spent(feature: str, limit: int) -> None: """Say that a request was refused for the per-user daily cap. The 429 is in the request line either way; which of the caps produced it is not, and they are answered by different things — @@ -189,20 +200,36 @@ def search_youtube( # Channels this user has blocked are dropped before ingestion — never shown, never stored. blocked = queries.blocked_ids(db, user) - # Whether search.list was called: YouTube charges a call it refuses too (the client records the - # units before it reads the status), so from then on the search has spent the shared budget. + # Whether search.list was paid for: YouTube charges a call it refuses too, so once the client + # recorded its units — or the request went out and no answer came back — the search has spent + # the shared budget. One that never reached YouTube (no connection, no token) has not. charged = False def fetch(token: str | None) -> dict: - nonlocal charged if source != "api": return search_scrape.search(term, continuation=token) - with _api_search, quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH): + if not _api_search.acquire(timeout=API_SEARCH_WAIT_SECONDS): + _busy() + raise refused(429, "search_busy") + try: + return paid_search(token) + finally: + _api_search.release() + + def paid_search(token: str | None) -> dict: + nonlocal charged + with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH): if not quota.can_spend(db, SEARCH_COST): _budget_spent() raise refused(429, "quota_spent") + spent = yt.spent + try: + page = yt.search_videos(term, page_token=token) + except YouTubeError as exc: + charged = yt.spent > spent or exc.sent + raise charged = True - return yt.search_videos(term, page_token=token) + return page def candidates_of(items: list[dict], _collected: list[str]) -> list[dict]: # Drop currently-live/upcoming, channel-less, and blocked-channel results. @@ -262,7 +289,8 @@ def search_youtube( except Exception: # A search that showed the user nothing does not count against them, unless search.list # was already paid for: then the slot is what stops a failing search (a bogus cursor gets - # a 400 every time) from spending the shared budget without end. + # a 400 every time) from spending the shared budget without end. A busy refusal paid + # nothing and gives it back. if not charged: _give_back(claimed, "search", user.id) raise diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index b7c549e0..8cb188ec 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -30,12 +30,26 @@ class YouTubeError(Exception): message is prose, it changes under us, and it is deliberately the poorer of the two here. `path` is the API endpoint that ANSWERED with the failure (`playlistItems`); empty when no - endpoint did — the request never got through (`transport`) or the token step refused first.""" + endpoint did — the request never got through (`transport`) or the token step refused first. - def __init__(self, message: str, *, reason: str = "", path: str = ""): + `sent` is a `transport` failure that came after the request may have gone out (a read timeout, + a dropped answer): nothing answered, but YouTube may have counted the call.""" + + def __init__(self, message: str, *, reason: str = "", path: str = "", sent: bool = False): super().__init__(message) self.reason = reason self.path = path + self.sent = sent + + +# The transport failures that stop a request before it reaches YouTube: nothing went out. +_UNSENT = ( + httpx.ConnectError, + httpx.ConnectTimeout, + httpx.PoolTimeout, + httpx.ProxyError, + httpx.UnsupportedProtocol, +) def _error_reason(resp: httpx.Response) -> str: @@ -162,6 +176,9 @@ def best_thumbnail(thumbnails: dict | None) -> str | None: class YouTubeClient: + # The quota units this client has recorded: what its calls have spent of the shared budget. + spent: int = 0 + def __init__(self, db, user: User): self.db = db self.user = user @@ -189,9 +206,16 @@ class YouTubeClient: return self._http.request(method, url, **kwargs) except httpx.HTTPError as exc: raise YouTubeError( - f"YouTube API unreachable ({exc.__class__.__name__}: {exc})", reason="transport" + f"YouTube API unreachable ({exc.__class__.__name__}: {exc})", + reason="transport", + sent=not isinstance(exc, _UNSENT), ) from exc + def _spend(self, units: int) -> None: + """Record what a call YouTube answered cost — a refused one too: it is charged all the same.""" + quota.record_usage(units) + self.spent += units + # --- auth --- def _access_token(self) -> str: tok = self.user.token @@ -269,7 +293,7 @@ class YouTubeClient: else: headers["Authorization"] = f"Bearer {self._access_token()}" resp = self._send("GET", f"{API_BASE}/{path}", params=p, headers=headers) - quota.record_usage(cost) + self._spend(cost) if resp.status_code != 200: raise _upstream_failure("GET", path, resp, expected) return resp.json() @@ -401,7 +425,7 @@ class YouTubeClient: params={"id": subscription_id}, headers={"Authorization": f"Bearer {self._access_token()}"}, ) - quota.record_usage(50) + self._spend(50) if resp.status_code not in (200, 204): raise _upstream_failure("DELETE", "subscriptions", resp) @@ -431,7 +455,7 @@ class YouTubeClient: def _write(self, method: str, path: str, *, params: dict, json: dict | None = None) -> dict: headers = {"Authorization": f"Bearer {self._access_token()}"} resp = self._send(method, f"{API_BASE}/{path}", params=params, json=json, headers=headers) - quota.record_usage(50) + self._spend(50) if resp.status_code not in (200, 204): raise _upstream_failure(method, path, resp) return resp.json() if resp.content else {} diff --git a/backend/tests/test_search_cap.py b/backend/tests/test_search_cap.py index 01c5661f..e79df0ba 100644 --- a/backend/tests/test_search_cap.py +++ b/backend/tests/test_search_cap.py @@ -6,8 +6,11 @@ outright gives its slot back; a zero limit turns live search off, which says so again tomorrow"; a lookup that fails says why instead of answering "no results"; only a few failures an hour get their slot back; and the API source — 100 units a page — reads one page per request, is refused before the call when the shared budget cannot pay for it (checked again under -one lock with the call), counts one slot per search though search.list logs its own event, and -keeps the slot of a search search.list was already paid for.""" +one lock with the call, whose wait is bounded), counts one slot per search though search.list logs +its own event, and keeps the slot of a search search.list was already paid for — but not of one +that never reached YouTube.""" + +import threading import pytest @@ -35,7 +38,7 @@ class _FakeYT: fail_videos = False def __init__(self, *_a, **_kw): - pass + self.spent = 0 def __enter__(self): return self @@ -47,14 +50,17 @@ class _FakeYT: return [] # search.list: the pages the test hands out, each logged as the real client does (100 units, - # under the action the route attributes it to). + # under the action the route attributes it to) — a refusal YouTube answered too, but not a + # failure no endpoint answered (no connection, no token). api_pages: list = [] api_calls = 0 def search_videos(self, q, page_token=None, order="relevance"): _FakeYT.api_calls += 1 - quota.record_usage(100) page = _FakeYT.api_pages.pop(0) + if not (isinstance(page, YouTubeError) and not page.path): + quota.record_usage(100) + self.spent += 100 if isinstance(page, Exception): raise page return page @@ -249,27 +255,98 @@ def test_an_api_search_the_budget_refused_at_the_call_gives_its_slot_back( def test_the_budget_check_and_the_call_it_pays_for_hold_one_lock( db, make_user, as_user, pages, monkeypatch ): - # Checked apart, searches sent together at the last 100 units all passed, and all spent. + # Checked apart, searches sent together at the last 100 units all passed, and all spent: the + # check and the call sit inside ONE hold of the lock, not two. user = make_user() _api(db) - held: list[tuple[str, bool]] = [] + events: list[str] = [] real_check = quota.can_spend real_call = _FakeYT.search_videos + class _Recorded: + def __init__(self): + self._lock = threading.Lock() + + def acquire(self, timeout=-1): + events.append("acquire") + return self._lock.acquire(timeout=timeout) + + def release(self): + events.append("release") + self._lock.release() + def can_spend(db_, units): - held.append(("check", search_routes._api_search.locked())) + events.append("check") return real_check(db_, units) def search_videos(self, q, page_token=None, order="relevance"): - held.append(("call", search_routes._api_search.locked())) + events.append("call") return real_call(self, q, page_token=page_token, order=order) + monkeypatch.setattr(search_routes, "_api_search", _Recorded()) monkeypatch.setattr(search_routes.quota, "can_spend", can_spend) monkeypatch.setattr(_FakeYT, "search_videos", search_videos) _FakeYT.api_pages = [_page(1)] assert as_user(user).get("/api/search/youtube?q=cats").status_code == 200 - assert held == [("check", False), ("check", True), ("call", True)] - assert not search_routes._api_search.locked() + assert events == ["check", "acquire", "check", "call", "release"] + + +def test_a_search_kept_waiting_for_the_lock_is_refused_as_busy_and_keeps_nothing( + db, make_user, as_user, pages, monkeypatch +): + # Each waiter holds a request worker: a slow YouTube must not park them all behind one another. + user = make_user() + _api(db) + monkeypatch.setattr(search_routes, "API_SEARCH_WAIT_SECONDS", 0.05) + _FakeYT.api_pages = [_page(1)] + # Held by another search for a while: a wait without its bound would get it, and be answered. + assert search_routes._api_search.acquire(timeout=1) + later = threading.Timer(2.0, search_routes._api_search.release) + later.start() + try: + resp = as_user(user).get("/api/search/youtube?q=cats") + finally: + later.join() + assert resp.status_code == 429 + assert resp.json()["detail"]["reason"] == "search_busy" + assert _FakeYT.api_calls == 0 + assert quota.actions_today(db, user.id, ACTION) == 0 + + +@pytest.mark.parametrize( + "failure", + [ + YouTubeError("YouTube API unreachable (ConnectError)", reason="transport"), + YouTubeError("User has no stored OAuth token"), + ], + ids=["no-connection", "no-token"], +) +def test_an_api_search_that_never_reached_youtube_gives_its_slot_back( + db, make_user, as_user, pages, failure +): + user = make_user() + _api(db) + sysconfig.set_value(db, "search_daily_limit_per_user", 1) + _FakeYT.api_pages = [failure, _page(1)] + client = as_user(user) + assert client.get("/api/search/youtube?q=cats").status_code == 422 + assert quota.actions_today(db, user.id, ACTION) == 0 + assert client.get("/api/search/youtube?q=cats").status_code == 200 + + +def test_an_api_search_whose_answer_never_came_keeps_its_slot(db, make_user, as_user, pages): + # The request went out (a read timeout): YouTube may have counted it. + user = make_user() + _api(db) + sysconfig.set_value(db, "search_daily_limit_per_user", 1) + _FakeYT.api_pages = [ + YouTubeError("YouTube API unreachable (ReadTimeout)", reason="transport", sent=True), + _page(1), + ] + client = as_user(user) + assert client.get("/api/search/youtube?q=cats").status_code == 422 + assert quota.actions_today(db, user.id, ACTION) == 1 + assert client.get("/api/search/youtube?q=cats").status_code == 429 def test_failures_past_the_hours_refunds_keep_their_slot(db, make_user, as_user, pages): diff --git a/backend/tests/test_youtube_refusals.py b/backend/tests/test_youtube_refusals.py index 94d51dbd..2400578a 100644 --- a/backend/tests/test_youtube_refusals.py +++ b/backend/tests/test_youtube_refusals.py @@ -193,3 +193,45 @@ def test_any_other_token_refusal_keeps_the_grant(monkeypatch): c._access_token() assert tok.scopes == "s" assert refusal_reason(caught.value) == "unavailable" + + +@pytest.mark.parametrize( + ("failure", "sent"), + [ + (httpx.ConnectError("refused"), False), + (httpx.ConnectTimeout("slow connect"), False), + (httpx.ProxyError("proxy down"), False), + (httpx.ReadTimeout("no answer"), True), + (httpx.RemoteProtocolError("dropped"), True), + ], +) +def test_a_transport_failure_says_whether_the_request_went_out(failure, sent): + """A search that never reached YouTube spent nothing and gets its slot back; one whose answer + never came may have been counted, and keeps it (`routes/search`).""" + + def request(*_a, **_k): + raise failure + + c = yt.YouTubeClient.__new__(yt.YouTubeClient) + c._http = SimpleNamespace(request=request) + with pytest.raises(YouTubeError) as caught: + c._send("GET", "https://example.invalid") + assert caught.value.reason == "transport" + assert caught.value.sent is sent + + +def test_a_call_youtube_answered_is_spent_a_refused_one_too(monkeypatch): + resp = httpx.Response( + 400, + json={"error": {"errors": [{"reason": "invalidPageToken"}]}}, + request=httpx.Request("GET", f"{yt.API_BASE}/search"), + ) + recorded: list[int] = [] + c = yt.YouTubeClient.__new__(yt.YouTubeClient) + c.db, c.spent = None, 0 + monkeypatch.setattr(yt.sysconfig, "get_str", lambda d, key: "a-key") + monkeypatch.setattr(yt.quota, "record_usage", recorded.append) + monkeypatch.setattr(c, "_send", lambda *a, **k: resp) + with pytest.raises(YouTubeError): + c._get("search", {}, cost=100) + assert (recorded, c.spent) == ([100], 100) diff --git a/frontend/src/i18n/locales/en/errors.json b/frontend/src/i18n/locales/en/errors.json index 1b6c9341..e9e86c90 100644 --- a/frontend/src/i18n/locales/en/errors.json +++ b/frontend/src/i18n/locales/en/errors.json @@ -190,6 +190,7 @@ "registration_closed": "Registration is currently closed", "related_disabled": "Related videos are turned off on this instance.", "related_daily_limit": "You've reached today's related videos limit ({{limit}}). Opening a list counts as one request, and so does every “Show more”. Try again tomorrow.", + "search_busy": "Too many searches are running right now. Try again in a few seconds.", "search_disabled": "Live YouTube search is turned off on this instance.", "search_daily_limit": "You've reached today's YouTube search limit ({{limit}}). Try again tomorrow.", "search_term_required": "Enter something to search for", diff --git a/frontend/src/i18n/locales/hu/errors.json b/frontend/src/i18n/locales/hu/errors.json index 09452956..d09b4406 100644 --- a/frontend/src/i18n/locales/hu/errors.json +++ b/frontend/src/i18n/locales/hu/errors.json @@ -190,6 +190,7 @@ "registration_closed": "A regisztráció jelenleg zárva van", "related_disabled": "A kapcsolódó videók ki vannak kapcsolva ezen a példányon.", "related_daily_limit": "Elérted a kapcsolódó videók mai korlátját ({{limit}}). Egy lista megnyitása és minden „Mutass többet” egy-egy kérésnek számít. Próbáld holnap.", + "search_busy": "Most túl sok keresés fut egyszerre. Próbáld újra pár másodperc múlva.", "search_disabled": "Az élő YouTube-keresés ki van kapcsolva ezen a példányon.", "search_daily_limit": "Elérted a mai YouTube-keresési korlátot ({{limit}}). Próbáld holnap.", "search_term_required": "Írj be egy keresőkifejezést",