test(backfill): lock the per-page durability and reserve-guard properties

Deep-review round 1 found the new tests asserted end-state only, so both
safety mechanisms this change adds were untested (a regression would ship
green):

- Per-page cursor durability: add test_a_mid_run_failure_leaves_the_prior_page
  _durably_committed, which fails the 2nd page's fetch after page 1 committed
  and asserts the cursor + rows are already persisted (re-read via db.expire).
  Mutation-verified: reverting to a single end-of-loop commit makes it fail
  while the pre-existing paging tests stay green — the property the original
  test could not distinguish.
- Reserve guard: add test_backfill_reserve_reached_tracks_the_boundary, driving
  real record_usage spend across the 9000/2000 default boundary and asserting
  the un-mocked quota.backfill_reserve_reached() flips at exactly remaining<=reserve.

Full backend suite: 244 passed, 2 skipped (network-gated).

backend/tests/test_channel_backfill.py

@@ -3,6 +3,7 @@ the update_channel -> immediate-backfill wiring (incl. the demo guard), and the
 "get full history" guard that keeps it from racing the scheduler's cursor or draining the budget."""
 from datetime import datetime, timezone
 
+import pytest
 from fastapi.testclient import TestClient
 
 import app.routes.channels as channels_route
@@ -207,6 +208,22 @@ class _PagedYt:
         return self.pages[page_token]
 
 
+class _RaisingYt:
+    """Serves pages like _PagedYt but raises when asked for `raise_on_token`, to simulate a
+    network/API failure that lands BETWEEN pages of a deep backfill."""
+
+    def __init__(self, pages, raise_on_token):
+        self.pages = pages
+        self.raise_on_token = raise_on_token
+        self.calls: list[str | None] = []
+
+    def get_playlist_items(self, playlist_id, page_token=None):
+        self.calls.append(page_token)
+        if page_token == self.raise_on_token:
+            raise RuntimeError("simulated mid-backfill failure")
+        return self.pages[page_token]
+
+
 class TestDeepBackfillPaging:
     """backfill_channel_deep must commit the advanced cursor after every page (durable resume, no
     re-charging quota for pages already stored) and yield the moment the shared budget hits the
@@ -276,3 +293,26 @@ class TestDeepBackfillPaging:
         assert yt.calls == [None]  # stopped before fetching the second page
         assert ch.backfill_cursor == "t1"
         assert ch.backfill_done is False
+
+    def test_a_mid_run_failure_leaves_the_prior_page_durably_committed(self, db, monkeypatch):
+        # The whole point of committing per page (vs one commit after the loop): if paging blows up
+        # mid-run, the pages already ingested stay durable, so the next run resumes from the stored
+        # cursor instead of re-fetching — and re-charging quota for — them. A single end-of-loop
+        # commit would lose the advanced cursor on the crash (it never reaches the post-loop commit),
+        # so this asserts the property the per-page commit actually buys.
+        monkeypatch.setattr(videos.quota, "backfill_reserve_reached", lambda: False)  # budget ok
+        ch = self._channel(db)
+        yt = _RaisingYt(
+            {None: _playlist_page(["a1", "a2"], next_token="t1")},
+            raise_on_token="t1",  # page 1 succeeds; fetching page 2 (token 't1') fails
+        )
+
+        with pytest.raises(RuntimeError):
+            videos.backfill_channel_deep(db, yt, ch, max_pages=5)
+
+        # Page 1's cursor AND rows were committed before the page-2 crash — re-read from the DB
+        # (expire) to prove it's persisted, not just set on the in-memory object.
+        db.expire(ch)
+        assert ch.backfill_cursor == "t1"
+        assert ch.backfill_done is False
+        assert db.query(Video).filter(Video.channel_id == ch.id).count() == 2

backend/tests/test_quota_ownership.py

@@ -55,3 +55,15 @@ def test_log_action_survives_caller_rollback(db):
     assert events[0].units == 0
     # ... and the caller's uncommitted user is gone.
     assert db.scalar(select(func.count()).select_from(User)) == 1
+
+
+def test_backfill_reserve_reached_tracks_the_boundary(db):
+    # The shared yield gate for every backfill/explore/revalidation path: True once today's remaining
+    # budget is at or below the backfill reserve. It reads in its OWN session, so it must reflect
+    # record_usage's independently-committed spend — and the boundary is inclusive (<=). Defaults:
+    # quota_daily_budget 9000, backfill_quota_reserve 2000 -> reached once 7000 units are used.
+    assert quota.backfill_reserve_reached() is False  # 0 used, remaining 9000
+    quota.record_usage(6999)                          # remaining 2001, just above the reserve
+    assert quota.backfill_reserve_reached() is False
+    quota.record_usage(1)                             # remaining 2000, exactly at the reserve
+    assert quota.backfill_reserve_reached() is True   # boundary is <=, so the reserve is reached