fix(search): every related and live-search request takes a slot, and a failed one gives it back

A related list's cursor cost nothing, and a cursor is a free string: a replayed one read and
enriched without end. Every request now takes a slot of the user's daily cap — taken under a lock
before the work (`quota.claim_action`), so requests sent together cannot all pass the last free
one, and given back when the request fails outright (`quota.refund_action`). The live search's
scrape source takes its slots the same way.

A limit of 0 turns the feature off: a distinct 409 `related_disabled` / `search_disabled`, and
`/api/me` says `related_enabled` / `search_enabled` so the UI can leave it out.

A videos.list that fails no longer reads as an empty page: `ingest_candidates` raises, and the
search and the related list answer the refusal (the shared budget spent says "quota") instead of
"no results". The two routes' page loops are one (`_gather`).

`clear` deletes only what the caller found: a related list's rows carry no find, and any user
could name them.

backend/tests/test_related_route.py

@@ -6,16 +6,18 @@ watched or hid — here or in their Takeout history, unless taken back to "new"
 channels and the video itself; it pages on until it has enough; it lands in the catalog the way
 search finds do; and a failure, or the day's cap, says so instead of answering an empty list."""
 
+import threading
 from datetime import datetime, timedelta, timezone
 
 import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select, text
 
 from app import quota, sysconfig
 from app.catalog.models import BlockedChannel, Channel, Subscription, Video
 from app.models import QuotaEvent
 from app.reader.models import TakeoutItem, TakeoutKind, VideoState, VideoUnmark
 from app.routes import search as search_routes
+from app.youtube.client import YouTubeError
 from app.youtube.innertube import ScrapeError
 
 T0 = datetime(2026, 10, 1, tzinfo=timezone.utc)
@@ -153,6 +155,8 @@ def test_what_the_takeout_history_shows_watched_is_left_out_unless_taken_back(
     db.commit()
     pages["pages"] = [_page([_item(n) for n in range(1, 5)], token=None)]
     assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(2), _vid(3), _vid(4)]
+    # Left out before the ingest: what the user has seen is not stored for a list that hides it.
+    assert db.get(Video, _vid(1)) is None
 
 
 def test_an_unmark_older_than_the_history_row_does_not_bring_it_back(db, make_user, as_user, pages):
@@ -288,14 +292,108 @@ def test_one_list_is_one_event_against_the_days_cap(db, make_user, as_user, page
     assert len(pages["calls"]) == 2  # the refused one asked YouTube nothing
 
 
-def test_reading_further_is_the_same_list_not_another_one(db, make_user, as_user, pages):
+def test_reading_further_takes_a_slot_too(db, make_user, as_user, pages):
+    # Each request scrapes and enriches; a cursor is a free string, so one that cost nothing let a
+    # replayed cursor read without end.
     user = make_user()
-    sysconfig.set_value(db, "related_daily_limit_per_user", 2)
+    sysconfig.set_value(db, "related_daily_limit_per_user", 1)
     pages["pages"] = [_page([_item(1)], token=None), _page([_item(2)], token=None)]
     client = as_user(user)
     assert client.get(f"/api/search/related/{SOURCE}").status_code == 200
-    assert client.get(f"/api/search/related/{SOURCE}?cursor=more").status_code == 200
-    assert quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_RELATED) == 1
+    refused = client.get(f"/api/search/related/{SOURCE}?cursor=more")
+    assert refused.status_code == 429
+    assert refused.json()["detail"]["reason"] == "related_daily_limit"
+    assert len(pages["calls"]) == 1
+
+
+def test_a_list_that_failed_gives_its_slot_back(db, make_user, as_user, pages):
+    user = make_user()
+    sysconfig.set_value(db, "related_daily_limit_per_user", 1)
+    pages["pages"] = [ScrapeError("blocked"), _page([_item(1)], token=None)]
+    client = as_user(user)
+    assert client.get(f"/api/search/related/{SOURCE}").status_code == 422
+    assert quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_RELATED) == 0
+    assert _ids(client.get(f"/api/search/related/{SOURCE}")) == [_vid(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)
+    resp = as_user(user).get(f"/api/search/related/{SOURCE}")
+    assert resp.status_code == 409
+    assert resp.json()["detail"] == {"code": "refused", "reason": "related_disabled"}
+    assert pages["calls"] == []
+
+
+def test_the_last_slot_is_taken_under_a_lock(db, make_user):
+    # Checked first and logged after seconds of scraping, requests sent together at the last free
+    # slot all passed. The claim waits for one already taking a slot for this user and action.
+    user = make_user()
+    action = quota.QuotaAction.VIDEOS_RELATED
+    db.execute(select(func.pg_advisory_xact_lock(func.hashtext(action), user.id)))
+    got: list[int | None] = []
+    t = threading.Thread(target=lambda: got.append(quota.claim_action(user.id, action, 1)))
+    t.start()
+    t.join(timeout=1)
+    assert t.is_alive()  # it waits for the one holding the lock
+    db.commit()
+    t.join(timeout=10)
+    assert got and got[0] is not None
+    assert quota.claim_action(user.id, action, 1) is None  # the one slot is taken
+
+
+def _videos_list_fails(monkeypatch, *, on_call: int = 1, reason: str = "quotaExceeded") -> None:
+    """YouTube's videos.list refuses from its `on_call`-th call on."""
+    calls = {"n": 0}
+    real = _FakeYT.get_videos
+
+    def get_videos(self, ids):
+        calls["n"] += 1
+        if calls["n"] >= on_call:
+            raise YouTubeError("videos -> 403", reason=reason, path="videos")
+        return real(self, ids)
+
+    monkeypatch.setattr(_FakeYT, "get_videos", get_videos)
+
+
+def test_a_lookup_that_fails_says_why_not_no_videos(db, make_user, as_user, pages, monkeypatch):
+    # Nothing it found can be kept unenriched; an empty list would read as "nothing related".
+    user = make_user()
+    _videos_list_fails(monkeypatch)
+    pages["pages"] = [_page([_item(1)], token=None)]
+    resp = as_user(user).get(f"/api/search/related/{SOURCE}")
+    assert resp.status_code == 422
+    assert resp.json()["detail"] == {"code": "youtube", "reason": "quota"}
+    assert quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_RELATED) == 0
+
+
+def test_a_lookup_failing_on_a_later_page_keeps_what_came_before(
+    db, make_user, as_user, pages, monkeypatch
+):
+    user = make_user()
+    _videos_list_fails(monkeypatch, on_call=2)
+    pages["pages"] = [_page([_item(1)], token="page2"), _page([_item(2)], token=None)]
+    assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(1)]
+
+
+def test_an_ingest_that_breaks_the_session_is_rolled_back(
+    db, make_user, as_user, pages, monkeypatch
+):
+    # A database error leaves the session unusable until it is rolled back: the rows of the pages
+    # before could not be read for the answer.
+    user = make_user()
+    real = search_routes.catalog_search.ingest_candidates
+    calls = {"n": 0}
+
+    def ingest(db_, yt, user_, candidates):
+        calls["n"] += 1
+        if calls["n"] == 2:
+            db_.execute(text("SELECT 1/0"))
+        return real(db_, yt, user_, candidates)
+
+    monkeypatch.setattr(search_routes.catalog_search, "ingest_candidates", ingest)
+    pages["pages"] = [_page([_item(1)], token="page2"), _page([_item(2)], token=None)]
+    assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(1)]
 
 
 def test_each_row_says_whether_the_user_follows_its_channel(db, make_user, as_user, pages):