fix(search): a paid search keeps its slot, refunds are few, and the budget check holds the call

An API-sourced live search that reached search.list has spent 100 units (the client records them
before it reads the status), so it keeps its daily slot: refunding it let a bogus cursor spend the
shared budget without ever counting against the cap. A failure before the call still gives the slot
back.

Every refund is rationed too (5 an hour per user and list): a cursor is a free string, and a request
built to fail would otherwise scrape YouTube from the server without end.

The API source asks the budget again right before search.list, under one process lock with the
call: checked apart, searches sent together at the last 100 units all passed and all spent.

backend/app/routes/search.py

@@ -20,6 +20,7 @@ cap — with what this user has already seen left out.
 This is the HTTP side — the paging, the caps and the refusals; what a page of results leaves in the
 catalog is `catalog/search`."""
 
+import threading
 from collections.abc import Callable
 from typing import Literal
 
@@ -33,6 +34,7 @@ from app.catalog import queries
 from app.catalog import search as catalog_search
 from app.db import get_db
 from app.models import User
+from app.ratelimit import RateLimiter
 from app.reader import queries as reader_queries
 from app.youtube import related_scrape, search_scrape
 from app.youtube.innertube import ScrapeError
@@ -55,6 +57,16 @@ RELATED_MAX_PAGES = 4
 # The locale YouTube fills a related list in: on the video's topic, or with what trends in Hungary.
 RELATED_REGIONS = {"topical": ("en", "US"), "hu": ("hu", "HU")}
 
+# A request that fails outright gives its slot back, but its shape is the user's (a cursor is a
+# free string): one built to fail would read YouTube for nothing, without end. So only this many
+# refunds an hour per user and list; past them a failure keeps its slot, as a success does.
+REFUNDS_PER_HOUR = 5
+_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.
+_api_search = threading.Lock()
+
 
 def _gather(
     db: Session,
@@ -106,6 +118,25 @@ def _gather(
     return collected, next_cursor
 
 
+def _give_back(claimed: int, feature: str, user_id: int) -> None:
+    """Refund the slot of a request that failed outright, while this user has refunds left."""
+    if _refunds.allow(f"{feature}:{user_id}"):
+        quota.refund_action(claimed)
+        return
+    log.info(
+        "a failed request kept its slot: this hour's refunds are spent",
+        extra=logs.event(f"{feature}.refund_refused", per_hour=REFUNDS_PER_HOUR),
+    )
+
+
+def _budget_spent() -> None:
+    """Say that a live search was refused because the shared YouTube budget cannot pay for it."""
+    log.info(
+        "a live search was refused: the shared YouTube budget is spent",
+        extra=logs.event("search.refused", reason="shared_quota", cost=SEARCH_COST),
+    )
+
+
 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 —
@@ -142,12 +173,10 @@ def search_youtube(
     daily_limit = sysconfig.get_int(db, "search_daily_limit_per_user")
     if daily_limit <= 0:
         raise refused(409, "search_disabled")
-    # The 100-unit budget pre-check only applies to the API source.
+    # The 100-unit budget pre-check only applies to the API source. It is asked again under the
+    # lock, right before the call (`fetch`); this one refuses before a slot is taken.
     if source == "api" and not quota.can_spend(db, SEARCH_COST):
-        log.info(
-            "a live search was refused: the shared YouTube budget is spent",
-            extra=logs.event("search.refused", reason="shared_quota", cost=SEARCH_COST),
-        )
+        _budget_spent()
         raise refused(429, "quota_spent")
     # The check and the take are one step (`quota.claim_action`), whichever the source: checked
     # first and logged after, requests sent together at the last free slot all passed — and with
@@ -160,11 +189,20 @@ 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.
+    charged = False
+
     def fetch(token: str | None) -> dict:
-        if source == "api":
-            with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
-                return yt.search_videos(term, page_token=token)
+        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 quota.can_spend(db, SEARCH_COST):
+                _budget_spent()
+                raise refused(429, "quota_spent")
+            charged = True
+            return yt.search_videos(term, page_token=token)
 
     def candidates_of(items: list[dict], _collected: list[str]) -> list[dict]:
         # Drop currently-live/upcoming, channel-less, and blocked-channel results.
@@ -222,8 +260,11 @@ def search_youtube(
                 ingest_failed=ingest_failed,
             )
     except Exception:
-        # A search that showed the user nothing does not count against them.
-        quota.refund_action(claimed)
+        # 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.
+        if not charged:
+            _give_back(claimed, "search", user.id)
         raise
 
     ordered = collected[:limit]
@@ -281,7 +322,8 @@ def related_videos(
     # Every request takes a slot, a cursor's too: each one scrapes pages and enriches what they
     # hold, and a cursor is a free string — counting only first pages let a replayed cursor read
     # without end. The check and the take are one step (`quota.claim_action`); a request that fails
-    # outright gives its slot back. A limit of 0 turns related videos off on this instance.
+    # outright gives its slot back, a few an hour (`_give_back`). A limit of 0 turns related videos
+    # off on this instance.
     daily_limit = sysconfig.get_int(db, "related_daily_limit_per_user")
     if daily_limit <= 0:
         raise refused(409, "related_disabled")
@@ -327,7 +369,7 @@ def related_videos(
                 ingest_failed=ingest_failed,
             )
     except Exception:
-        quota.refund_action(claimed)
+        _give_back(claimed, "related", user.id)
         raise
 
     rows = reader_queries.feed_rows_by_ids(db, user, collected)

backend/tests/test_related_route.py

@@ -316,6 +316,18 @@ def test_a_list_that_failed_gives_its_slot_back(db, make_user, as_user, pages):
     assert _ids(client.get(f"/api/search/related/{SOURCE}")) == [_vid(1)]
 
 
+def test_failures_past_the_hours_refunds_keep_their_slot(db, make_user, as_user, pages):
+    # A cursor is a free string: a request built to fail would otherwise read YouTube without end.
+    user = make_user()
+    sysconfig.set_value(db, "related_daily_limit_per_user", 50)
+    n = search_routes.REFUNDS_PER_HOUR
+    pages["pages"] = [ScrapeError("bad cursor") for _ in range(n + 1)]
+    client = as_user(user)
+    for _ in range(n + 1):
+        assert client.get(f"/api/search/related/{SOURCE}?cursor=bogus").status_code == 422
+    assert quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_RELATED) == 1
+
+
 def test_a_zero_limit_turns_related_videos_off(db, make_user, as_user, pages):
     user = make_user()
     sysconfig.set_value(db, "related_daily_limit_per_user", 0)

backend/tests/test_search_cap.py

@@ -3,9 +3,11 @@
 The scrape and search.list are replaced by pages the test hands out, and YouTube's videos.list by an
 offline enrich. What is held: every request takes a slot, a cursor's too; a search that failed
 outright gives its slot back; a zero limit turns live search off, which says so instead of "try
-again tomorrow"; a lookup that fails says why instead of answering "no results"; 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, and counts one slot per search though search.list logs its own event."""
+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."""
 
 import pytest
 
@@ -210,15 +212,73 @@ def test_the_api_source_is_refused_before_the_call_when_the_budget_cannot_pay(
     assert quota.actions_today(db, user.id, ACTION) == 0
 
 
-def test_an_api_search_that_failed_gives_its_slot_back(db, make_user, as_user, pages):
+def test_an_api_search_that_search_list_refused_keeps_its_slot(db, make_user, as_user, pages):
+    # YouTube charges a refused call too (the client records the units first), so a search that
+    # reached search.list has spent the shared budget: refunding it let a bogus cursor spend it all.
     user = make_user()
     _api(db)
     sysconfig.set_value(db, "search_daily_limit_per_user", 1)
     _FakeYT.api_pages = [
-        YouTubeError("search -> 403", reason="quotaExceeded", path="search"),
+        YouTubeError("search -> 400", reason="invalidPageToken", path="search"),
         _page(1),
     ]
     client = as_user(user)
-    assert client.get("/api/search/youtube?q=cats").status_code == 422
+    assert client.get("/api/search/youtube?q=cats&cursor=bogus").status_code == 422
+    assert quota.actions_today(db, user.id, ACTION) == 1
+    refused = client.get("/api/search/youtube?q=cats")
+    assert refused.status_code == 429
+    assert refused.json()["detail"]["reason"] == "search_daily_limit"
+    assert _FakeYT.api_calls == 1
+
+
+def test_an_api_search_the_budget_refused_at_the_call_gives_its_slot_back(
+    db, make_user, as_user, pages, monkeypatch
+):
+    user = make_user()
+    _api(db)
+    answers = iter([True, False])  # the pre-check passes; the budget is gone by the call
+    monkeypatch.setattr(search_routes.quota, "can_spend", lambda _db, _u: next(answers))
+    _FakeYT.api_pages = [_page(1)]
+    resp = as_user(user).get("/api/search/youtube?q=cats")
+    assert resp.status_code == 429
+    assert resp.json()["detail"]["reason"] == "quota_spent"
+    assert _FakeYT.api_calls == 0
     assert quota.actions_today(db, user.id, ACTION) == 0
-    assert client.get("/api/search/youtube?q=cats").status_code == 200
+
+
+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.
+    user = make_user()
+    _api(db)
+    held: list[tuple[str, bool]] = []
+    real_check = quota.can_spend
+    real_call = _FakeYT.search_videos
+
+    def can_spend(db_, units):
+        held.append(("check", search_routes._api_search.locked()))
+        return real_check(db_, units)
+
+    def search_videos(self, q, page_token=None, order="relevance"):
+        held.append(("call", search_routes._api_search.locked()))
+        return real_call(self, q, page_token=page_token, order=order)
+
+    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()
+
+
+def test_failures_past_the_hours_refunds_keep_their_slot(db, make_user, as_user, pages):
+    # A cursor is a free string: a request built to fail would otherwise read YouTube without end.
+    user = make_user()
+    sysconfig.set_value(db, "search_daily_limit_per_user", 50)
+    n = search_routes.REFUNDS_PER_HOUR
+    pages["pages"] = [ScrapeError("bad cursor") for _ in range(n + 1)]
+    client = as_user(user)
+    for _ in range(n + 1):
+        assert client.get("/api/search/youtube?q=cats&cursor=bogus").status_code == 422
+    assert quota.actions_today(db, user.id, ACTION) == 1