refactor(backfill): centralize the reserve-yield guard and drop the per-page connection hold
Addresses the code-review findings on the immediate full-history change: - Extract quota.backfill_reserve_reached(): the remaining/reserve compare was open-coded at seven sites (recent/deep/immediate backfill, explore, rolling revalidation, and the new per-page guard). It reads in its OWN short-lived session (like record_usage/log_action), so a per-page guard no longer holds the caller's transaction open across the YouTube fetch that follows — the 400-page immediate run no longer pins a pooled connection idle-in-transaction through each network round-trip. Migrated all call sites; dropped the now-dead sysconfig imports in channels.py and maintenance.py. - Simplify the deep-backfill loop: channel.backfill_done = not page_token (done is False on entry) plus one break, replacing the doubled `if not page_token`. - Tests patch the backfill_reserve_reached seam directly instead of mocking remaining_today internals, removing the fragile dead-value budget iterator. Full backend suite green; ruff clean.
This commit is contained in:
@@ -118,6 +118,20 @@ def can_spend(db: Session, units: int) -> bool:
|
||||
return remaining_today(db) >= units
|
||||
|
||||
|
||||
def backfill_reserve_reached() -> bool:
|
||||
"""True when today's remaining budget is at or below the backfill reserve — the shared yield
|
||||
condition for every backfill/explore path (recent, deep, immediate full-history, explore, and
|
||||
rolling revalidation).
|
||||
|
||||
Reads in its OWN short-lived session (like record_usage / log_action), for two reasons: it
|
||||
replaces the remaining/reserve comparison that was open-coded at every call site, and — critically
|
||||
for the per-page guards in paging loops — a guard must not hold the caller's transaction open
|
||||
across the network fetch that follows it. Quota is committed independently of any caller, so a
|
||||
fresh session reads the same numbers."""
|
||||
with SessionLocal() as qs:
|
||||
return remaining_today(qs) <= sysconfig.get_int(qs, "backfill_quota_reserve")
|
||||
|
||||
|
||||
def log_action(user_id: int, action: str) -> None:
|
||||
"""Record a zero-cost action event so per-user daily caps still count it.
|
||||
|
||||
|
||||
@@ -9,7 +9,7 @@ from pydantic import BaseModel
|
||||
from sqlalchemy import func, select, update
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app import quota, sysconfig
|
||||
from app import quota
|
||||
from app.auth import current_user, require_human, require_write_scope
|
||||
from app.db import get_db
|
||||
from app.models import (
|
||||
@@ -358,7 +358,7 @@ def explore_channel(
|
||||
raise HTTPException(status_code=404, detail="Unknown channel")
|
||||
if _is_blocked(db, user, channel_id):
|
||||
raise HTTPException(status_code=403, detail="You've blocked this channel.")
|
||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||
if quota.backfill_reserve_reached():
|
||||
raise HTTPException(
|
||||
status_code=429,
|
||||
detail="The shared daily quota reserve is reached — try exploring later.",
|
||||
|
||||
@@ -25,7 +25,7 @@ from datetime import timedelta
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app import progress, quota, sysconfig
|
||||
from app import progress, quota
|
||||
from app.config import settings
|
||||
from app.models import Playlist, PlaylistItem, Video
|
||||
from app.notifications import create_notification
|
||||
@@ -188,7 +188,7 @@ def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict:
|
||||
now = _now()
|
||||
# Process in videos.list-sized pages so we commit incrementally and stop on low quota.
|
||||
while checked < batch:
|
||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||
if quota.backfill_reserve_reached():
|
||||
break
|
||||
page = (
|
||||
db.execute(
|
||||
@@ -228,7 +228,7 @@ def run_maintenance(db: Session) -> dict:
|
||||
phase1 = _recheck_flagged(db, yt)
|
||||
phase2 = (
|
||||
_revalidate_rolling(db, yt)
|
||||
if quota.remaining_today(db) > sysconfig.get_int(db, "backfill_quota_reserve")
|
||||
if not quota.backfill_reserve_reached()
|
||||
else {"checked": 0, "flagged": 0, "skipped": "low quota"}
|
||||
)
|
||||
return {
|
||||
|
||||
@@ -183,7 +183,7 @@ def run_recent_backfill(
|
||||
videos_added = 0
|
||||
with YouTubeClient(db, user) as yt:
|
||||
for channel in channels:
|
||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||
if quota.backfill_reserve_reached():
|
||||
break
|
||||
try:
|
||||
videos_added += backfill_channel_recent(db, yt, channel)
|
||||
@@ -229,7 +229,7 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -
|
||||
videos_added = 0
|
||||
with YouTubeClient(db, user) as yt:
|
||||
for channel in channels:
|
||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||
if quota.backfill_reserve_reached():
|
||||
break
|
||||
# Claim the channel (or skip it if an interactive backfill already owns it) so the two
|
||||
# never advance the same cursor concurrently — symmetric with backfill_channel_now.
|
||||
@@ -294,7 +294,7 @@ def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 400) ->
|
||||
channel = db.get(Channel, channel_id)
|
||||
if channel is None or channel.backfill_done or not channel.uploads_playlist_id:
|
||||
return
|
||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||
if quota.backfill_reserve_reached():
|
||||
return
|
||||
try:
|
||||
# Attribute the spend to the requesting user under the full-backfill action, so it
|
||||
|
||||
@@ -168,9 +168,8 @@ def backfill_channel_deep(
|
||||
return 0
|
||||
page_token = channel.backfill_cursor
|
||||
total = 0
|
||||
reserve = sysconfig.get_int(db, "backfill_quota_reserve")
|
||||
for _ in range(max_pages):
|
||||
if quota.remaining_today(db) <= reserve:
|
||||
if quota.backfill_reserve_reached():
|
||||
break
|
||||
data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
|
||||
rows = [
|
||||
@@ -183,8 +182,7 @@ def backfill_channel_deep(
|
||||
total += insert_stubs(db, rows)
|
||||
page_token = data.get("nextPageToken")
|
||||
channel.backfill_cursor = page_token
|
||||
if not page_token:
|
||||
channel.backfill_done = True
|
||||
channel.backfill_done = not page_token
|
||||
db.add(channel)
|
||||
db.commit()
|
||||
if not page_token:
|
||||
|
||||
@@ -227,7 +227,7 @@ class TestDeepBackfillPaging:
|
||||
return ch
|
||||
|
||||
def test_cursor_is_committed_each_page_and_a_later_run_resumes(self, db, monkeypatch):
|
||||
monkeypatch.setattr(videos.quota, "remaining_today", lambda _db: 10_000)
|
||||
monkeypatch.setattr(videos.quota, "backfill_reserve_reached", lambda: False) # budget ok
|
||||
ch = self._channel(db)
|
||||
yt = _PagedYt(
|
||||
{
|
||||
@@ -258,11 +258,11 @@ class TestDeepBackfillPaging:
|
||||
assert ch.backfill_cursor is None
|
||||
|
||||
def test_yields_when_the_shared_budget_hits_the_reserve(self, db, monkeypatch):
|
||||
# Budget is above the 2000 reserve for the first page, then drops to the reserve — the guard
|
||||
# is checked BEFORE each page, so the run fetches exactly one page and then yields, leaving
|
||||
# the tail for the paced scheduler (cursor parked, backfill_done False).
|
||||
budget = iter([10_000, 2_000, 2_000])
|
||||
monkeypatch.setattr(videos.quota, "remaining_today", lambda _db: next(budget))
|
||||
# Budget is fine for the first page, then hits the reserve — the guard is checked BEFORE each
|
||||
# page, so the run fetches exactly one page and then yields, leaving the tail for the paced
|
||||
# scheduler (cursor parked, backfill_done False).
|
||||
reached = iter([False, True])
|
||||
monkeypatch.setattr(videos.quota, "backfill_reserve_reached", lambda: next(reached))
|
||||
ch = self._channel(db)
|
||||
yt = _PagedYt(
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user