diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 8221a5fd..02ead226 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -267,14 +267,19 @@ _backfill_lock = threading.Lock() _MAX_CONCURRENT_IMMEDIATE = 2 -def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 60) -> None: +def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 400) -> None: """User-initiated "get full history": deep-backfill ONE channel to completion right now, in a background task with its OWN session, so the full catalogue lands in seconds instead of over the - paced scheduler's 10-minute ticks (max_pages=5/channel/tick). Quota-guarded, capped to - _MAX_CONCURRENT_IMMEDIATE interactive runs so one user can't drain the shared backfill budget by - opting many channels in at once (the overflow, and any very large channel's tail, is finished by - the scheduler since deep_requested stays set), and mutually excluded per-channel with the scheduler - via _backfill_in_progress. Best-effort — logs and swallows failures.""" + paced scheduler's 10-minute ticks (max_pages=5/channel/tick). max_pages is high enough to finish + any realistic channel in a single run (400 pages = 20 000 videos; e.g. BBC Earth's ~3400 = 69 + pages — the old 60-page cap stopped one page short of large channels, so their tail dribbled in + over the scheduler's 10-minute ticks and looked stuck). backfill_channel_deep re-checks the shared + budget before each page and yields at the reserve, so a high cap still can't drain the day's quota. + Quota-guarded, capped to _MAX_CONCURRENT_IMMEDIATE interactive runs so one user can't drain the + shared backfill budget by opting many channels in at once (the overflow, and any channel too large + to finish under the reserve, is finished by the scheduler since deep_requested stays set), and + mutually excluded per-channel with the scheduler via _backfill_in_progress. Best-effort — logs and + swallows failures.""" global _immediate_running with _backfill_lock: if channel_id in _backfill_in_progress or _immediate_running >= _MAX_CONCURRENT_IMMEDIATE: diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index 3b2f605f..8270d346 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -10,7 +10,7 @@ from sqlalchemy import and_, func, or_, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session -from app import progress, sysconfig +from app import progress, quota, sysconfig from app.models import LIVE_OR_UPCOMING, Channel, Video from app.titles import normalize_title from app.youtube.client import YouTubeClient, best_thumbnail @@ -156,12 +156,22 @@ def backfill_channel_recent(db: Session, yt: YouTubeClient, channel: Channel) -> def backfill_channel_deep( db: Session, yt: YouTubeClient, channel: Channel, max_pages: int = 5 ) -> int: - """Continue paging older uploads from the stored cursor (bounded per run).""" + """Continue paging older uploads from the stored cursor (bounded per run). + + Commits the advanced cursor after every page (not just at the end) so progress + is durable: a restart resumes from the last stored page instead of re-fetching — + and re-charging quota for — pages already ingested. Re-checks the shared budget + before each page and yields once it hits the backfill reserve, so a single large + channel can't drain the day's quota in one run even with a high max_pages; the + paced scheduler finishes the tail on later ticks (deep_requested stays set).""" if channel.backfill_done or not channel.uploads_playlist_id: 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: + break data = yt.get_playlist_items(channel.uploads_playlist_id, page_token) rows = [ s @@ -175,9 +185,10 @@ def backfill_channel_deep( channel.backfill_cursor = page_token if not page_token: channel.backfill_done = True + db.add(channel) + db.commit() + if not page_token: break - db.add(channel) - db.commit() return total diff --git a/backend/tests/test_channel_backfill.py b/backend/tests/test_channel_backfill.py index 8da7c72d..58198b86 100644 --- a/backend/tests/test_channel_backfill.py +++ b/backend/tests/test_channel_backfill.py @@ -11,7 +11,7 @@ from app.auth import current_user from app.db import get_db from app.main import app from app.models import Channel, Subscription, User, Video -from app.sync import runner +from app.sync import runner, videos _PUB = datetime(2020, 1, 1, tzinfo=timezone.utc) @@ -176,3 +176,103 @@ class _NullYt: def __exit__(self, *a): return False + + +def _playlist_page(ids, next_token=None): + """A playlistItems.list response shaped for stub_from_playlist_item.""" + page = { + "items": [ + { + "contentDetails": {"videoId": vid, "videoPublishedAt": "2020-01-01T00:00:00Z"}, + "snippet": {"title": vid}, + } + for vid in ids + ] + } + if next_token is not None: + page["nextPageToken"] = next_token + return page + + +class _PagedYt: + """Serves a fixed page-token -> page map and records the tokens it was asked for, so a test can + assert exactly how many pages a deep-backfill run fetched (and where it resumed from).""" + + def __init__(self, pages): + self.pages = pages + self.calls: list[str | None] = [] + + def get_playlist_items(self, playlist_id, page_token=None): + self.calls.append(page_token) + 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 + backfill reserve — the two properties that make a high immediate max_pages safe.""" + + @staticmethod + def _channel(db): + ch = Channel( + id="UC_deep", + title="Deep", + recent_synced_at=_PUB, + uploads_playlist_id="UU_deep", + backfill_done=False, + backfill_cursor=None, + ) + db.add(ch) + db.commit() + 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) + ch = self._channel(db) + yt = _PagedYt( + { + None: _playlist_page(["a1", "a2"], next_token="t1"), + "t1": _playlist_page(["b1", "b2"], next_token="t2"), + "t2": _playlist_page(["c1"], next_token=None), # last page: no nextPageToken + } + ) + + # First run is bounded to 2 pages: it must stop mid-catalogue with the cursor durably parked + # on the next unfetched page and backfill_done still False. + added = videos.backfill_channel_deep(db, yt, ch, max_pages=2) + assert added == 4 + assert yt.calls == [None, "t1"] + assert ch.backfill_cursor == "t2" + assert ch.backfill_done is False + # Durably persisted (committed per page), not just set on the in-memory object. + db.expire(ch) + assert ch.backfill_cursor == "t2" + assert db.query(Video).filter(Video.channel_id == ch.id).count() == 4 + + # A later run resumes FROM the stored cursor (not from the start — no wasted re-fetch) and + # finishes: reaching the token-less last page flips backfill_done and clears the cursor. + added2 = videos.backfill_channel_deep(db, yt, ch, max_pages=5) + assert added2 == 1 + assert yt.calls == [None, "t1", "t2"] # resumed at "t2", did not re-fetch None/"t1" + assert ch.backfill_done is True + 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)) + ch = self._channel(db) + yt = _PagedYt( + { + None: _playlist_page(["a1", "a2"], next_token="t1"), + "t1": _playlist_page(["b1"], next_token="t2"), + } + ) + + added = videos.backfill_channel_deep(db, yt, ch, max_pages=50) + assert added == 2 + assert yt.calls == [None] # stopped before fetching the second page + assert ch.backfill_cursor == "t1" + assert ch.backfill_done is False diff --git a/frontend/src/components/ChannelPage.tsx b/frontend/src/components/ChannelPage.tsx index ca420456..b3516082 100644 --- a/frontend/src/components/ChannelPage.tsx +++ b/frontend/src/components/ChannelPage.tsx @@ -422,13 +422,18 @@ export default function ChannelPage({ )} {/* Only the recent subset is stored → offer to pull the channel's full back-catalogue. - Once queued (deep_requested) it shows as loading; hidden once the backfill is done. */} + Once queued (deep_requested) it shows as loading; hidden once the backfill is done. + While loading, the label carries the live stored/total count (the detail poll refreshes + it) so a large channel's paced tail reads as visible progress rather than a stuck + spinner. */} {ch?.subscribed && !ch.backfill_done && !me.is_demo && ( )} diff --git a/frontend/src/i18n/locales/en/channel.json b/frontend/src/i18n/locales/en/channel.json index b6bab16a..56f1ec5f 100644 --- a/frontend/src/i18n/locales/en/channel.json +++ b/frontend/src/i18n/locales/en/channel.json @@ -29,6 +29,8 @@ "showInFeed": "Show in my feed", "fullHistory": "Full history", "fullHistoryLoading": "Fetching history…", + "fullHistoryProgress": "Fetching history… {{stored}} / {{total}}", + "fullHistoryLoadingHint": "Fetching the full back-catalogue — videos appear as they're found; a very large channel's tail is finished in the background.", "fullHistoryQueued": "Fetching this channel's full history — new videos will appear as they're found.", "fullHistoryFailed": "Couldn't start the full-history fetch." } diff --git a/frontend/src/i18n/locales/hu/channel.json b/frontend/src/i18n/locales/hu/channel.json index ddaee097..814903b6 100644 --- a/frontend/src/i18n/locales/hu/channel.json +++ b/frontend/src/i18n/locales/hu/channel.json @@ -29,6 +29,8 @@ "showInFeed": "Mutasd a hírfolyamomban", "fullHistory": "Teljes előzmény", "fullHistoryLoading": "Előzmény letöltése…", + "fullHistoryProgress": "Előzmény letöltése… {{stored}} / {{total}}", + "fullHistoryLoadingHint": "A teljes előzmény letöltése folyamatban — a videók feltűnnek, ahogy előkerülnek; egy nagyon nagy csatorna maradékát a háttérben fejezi be.", "fullHistoryQueued": "A csatorna teljes előzményének letöltése elindult — az új videók feltűnnek, ahogy előkerülnek.", "fullHistoryFailed": "Nem sikerült elindítani a teljes előzmény letöltését." }