diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py index 2a29c6ec..70e62c69 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -477,7 +477,7 @@ def update_channel( # request returns right away; the channel page's live poll shows the count climb. The scheduler # still finishes a very large channel's tail on its normal runs (deep_requested stays set). if deep_turned_on: - background.add_task(backfill_channel_now, channel_id) + background.add_task(backfill_channel_now, channel_id, user.id) return { "id": channel_id, diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 4849daf9..8221a5fd 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -231,8 +231,12 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) - for channel in channels: if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"): break - if channel.id in _immediate_backfill: - continue # an interactive backfill_channel_now is paging this one — no cursor race + # 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. + with _backfill_lock: + if channel.id in _backfill_in_progress: + continue + _backfill_in_progress.add(channel.id) try: videos_added += backfill_channel_deep(db, yt, channel, max_pages) processed += 1 @@ -240,6 +244,9 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) - except Exception: db.rollback() log.exception("Deep backfill failed for channel %s", channel.id) + finally: + with _backfill_lock: + _backfill_in_progress.discard(channel.id) reconciled = reconcile_full_history(db) return { "channels_processed": processed, @@ -248,27 +255,32 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) - } -# Channels an interactive "get full history" is backfilling RIGHT NOW (backfill_channel_now runs in a -# FastAPI BackgroundTasks threadpool thread; the scheduler runs in its own thread — hence the lock). -# The scheduler's deep pass skips these so the two never page the same channel's cursor concurrently, -# and the size cap bounds how much immediate backfill one burst can kick, letting the rest fall back -# to the scheduler's fair, quota-paced runs. -_immediate_backfill: set[str] = set() -_immediate_lock = threading.Lock() +# Channels being deep-backfilled RIGHT NOW by EITHER path — the interactive backfill_channel_now (a +# FastAPI BackgroundTasks threadpool thread) OR the scheduler's run_deep_backfill (its own thread). +# BOTH register the channel they're paging (under the lock) so the other skips it: no two runs page +# the same channel's `backfill_cursor` concurrently, whichever started first. `_immediate_running` +# counts only the interactive path, capped so one user's burst can't drain the shared budget (the +# scheduler is already quota-paced). Single uvicorn process → one shared module state, lock suffices. +_backfill_in_progress: set[str] = set() +_immediate_running = 0 +_backfill_lock = threading.Lock() _MAX_CONCURRENT_IMMEDIATE = 2 -def backfill_channel_now(channel_id: str, max_pages: int = 60) -> None: +def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 60) -> 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, and capped to - _MAX_CONCURRENT_IMMEDIATE in flight 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 on its normal runs, since deep_requested stays set. Best-effort — logs and swallows.""" - with _immediate_lock: - if channel_id in _immediate_backfill or len(_immediate_backfill) >= _MAX_CONCURRENT_IMMEDIATE: - return # already running, or too many in flight — leave it to the scheduler - _immediate_backfill.add(channel_id) + 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.""" + global _immediate_running + with _backfill_lock: + if channel_id in _backfill_in_progress or _immediate_running >= _MAX_CONCURRENT_IMMEDIATE: + return # already backfilling this channel, or too many interactive runs — leave it to the scheduler + _backfill_in_progress.add(channel_id) + _immediate_running += 1 try: with SessionLocal() as db: user = get_service_user(db) @@ -280,15 +292,19 @@ def backfill_channel_now(channel_id: str, max_pages: int = 60) -> None: if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"): return try: - with YouTubeClient(db, user) as yt: - backfill_channel_deep(db, yt, channel, max_pages) - reconcile_full_history(db) + # Attribute the spend to the requesting user under the full-backfill action, so it + # shows up correctly in the usage breakdown (not the default OTHER/system bucket). + with quota.attribute(user_id, quota.QuotaAction.VIDEOS_BACKFILL_FULL): + with YouTubeClient(db, user) as yt: + backfill_channel_deep(db, yt, channel, max_pages) + reconcile_full_history(db) except Exception: db.rollback() log.exception("Immediate backfill failed for channel %s", channel_id) finally: - with _immediate_lock: - _immediate_backfill.discard(channel_id) + with _backfill_lock: + _backfill_in_progress.discard(channel_id) + _immediate_running -= 1 def estimate_deep_backfill(db: Session, channels: list[Channel]) -> dict: diff --git a/backend/tests/test_channel_backfill.py b/backend/tests/test_channel_backfill.py index a70a3296..8da7c72d 100644 --- a/backend/tests/test_channel_backfill.py +++ b/backend/tests/test_channel_backfill.py @@ -1,36 +1,37 @@ """R14 backend coverage: the channel-detail Normal/Shorts/Live counts (moved off the manager table), -and the interactive "get full history" immediate-backfill guard that keeps one burst from racing the -scheduler's cursor or draining the shared backfill budget.""" +the update_channel -> immediate-backfill wiring (incl. the demo guard), and the interactive +"get full history" guard that keeps it from racing the scheduler's cursor or draining the budget.""" from datetime import datetime, timezone from fastapi.testclient import TestClient +import app.routes.channels as channels_route import app.state as state from app.auth import current_user from app.db import get_db from app.main import app -from app.models import Channel, User, Video +from app.models import Channel, Subscription, User, Video from app.sync import runner +_PUB = datetime(2020, 1, 1, tzinfo=timezone.utc) + def test_channel_detail_returns_normal_shorts_live_counts(db, monkeypatch): # published_at set so channel_detail skips the YouTube-enrich path (no quota / network in tests). - ch = Channel( - id="UC_nsl", - title="N/S/L Channel", - published_at=datetime(2020, 1, 1, tzinfo=timezone.utc), - ) + ch = Channel(id="UC_nsl", title="N/S/L Channel", published_at=_PUB) db.add(ch) me = User(email="nsl@example.com") db.add(me) db.flush() - # 2 normal, 1 short, 1 live → stored 4, normal = 4 − 1 (short) − 1 (live) = 2. + # DISTINCT counts (2 shorts, 1 live) so a short<->live filter/field swap can't pass: + # stored 5, short 2, live 1, normal = 5 - 2 - 1 = 2. db.add_all( [ Video(id="v_norm1", channel_id=ch.id), Video(id="v_norm2", channel_id=ch.id), - Video(id="v_short", channel_id=ch.id, is_short=True), - Video(id="v_live", channel_id=ch.id, live_status="live"), + Video(id="v_short1", channel_id=ch.id, is_short=True), + Video(id="v_short2", channel_id=ch.id, is_short=True), + Video(id="v_live1", channel_id=ch.id, live_status="live"), ] ) db.commit() @@ -46,48 +47,99 @@ def test_channel_detail_returns_normal_shorts_live_counts(db, monkeypatch): assert resp.status_code == 200, resp.text body = resp.json() - assert body["stored_videos"] == 4 - assert body["count_short"] == 1 + assert body["stored_videos"] == 5 + assert body["count_short"] == 2 assert body["count_live"] == 1 assert body["count_normal"] == 2 +def _setup_channel_and_sub(db, *, email, is_demo=False): + ch = Channel( + id="UC_wire", + title="Wire", + published_at=_PUB, + recent_synced_at=_PUB, # so update_channel skips the in-request recent backfill (no network) + uploads_playlist_id="UU_wire", + ) + db.add(ch) + user = User(email=email, is_demo=is_demo) + db.add(user) + db.flush() + db.add(Subscription(user_id=user.id, channel_id=ch.id, deep_requested=False)) + db.commit() + return ch, user + + +def _patch_deep_requested(db, monkeypatch, user): + calls: list[tuple] = [] + # background.add_task runs after the response in TestClient; record what would be scheduled. + monkeypatch.setattr(channels_route, "backfill_channel_now", lambda *a, **k: calls.append(a)) + monkeypatch.setattr(state, "_configured_cache", True) + app.dependency_overrides[get_db] = lambda: db + app.dependency_overrides[current_user] = lambda: user + try: + resp = TestClient(app).patch("/api/channels/UC_wire", json={"deep_requested": True}) + finally: + app.dependency_overrides.clear() + return resp, calls + + +def test_deep_request_schedules_immediate_backfill_for_non_demo(db, monkeypatch): + ch, me = _setup_channel_and_sub(db, email="wire@example.com") + resp, calls = _patch_deep_requested(db, monkeypatch, me) + assert resp.status_code == 200, resp.text + assert resp.json()["deep_requested"] is True + # scheduled exactly once, with (channel_id, user_id) — the args backfill_channel_now needs. + assert calls == [(ch.id, me.id)] + + +def test_deep_request_does_not_schedule_immediate_backfill_for_demo(db, monkeypatch): + # The SHARED demo account must not be able to kick immediate full backfills (budget drain). + _ch, demo = _setup_channel_and_sub(db, email="demo@example.com", is_demo=True) + resp, calls = _patch_deep_requested(db, monkeypatch, demo) + assert resp.status_code == 200, resp.text + assert calls == [], "demo must not schedule an immediate backfill" + # …and deep_requested is not persisted on the shared demo subscription either. + assert resp.json()["deep_requested"] is False + + class TestImmediateBackfillGuard: - """backfill_channel_now coordinates with the paced scheduler through a small in-flight set: a - second call for a channel already in flight is a no-op, the set is capped so a burst can't drain - the shared budget, and run_deep_backfill skips a channel an immediate backfill is handling.""" + """backfill_channel_now coordinates with the paced scheduler through _backfill_in_progress (per- + channel mutual exclusion, both directions) + _immediate_running (a cap on interactive runs so a + burst can't drain the shared budget).""" def setup_method(self): - runner._immediate_backfill.clear() + runner._backfill_in_progress.clear() + runner._immediate_running = 0 def teardown_method(self): - runner._immediate_backfill.clear() + runner._backfill_in_progress.clear() + runner._immediate_running = 0 @staticmethod def _boom_session(): - # SessionLocal is only reached AFTER the guard admits the channel; if the guard short-circuits - # (as it must here) this is never called, so raising proves the short-circuit. + # Reached only AFTER the guard admits the channel; if the guard short-circuits (as it must + # in these tests) this never runs, so raising proves the short-circuit. raise AssertionError("guard must short-circuit before opening a session") def test_a_second_call_for_an_in_flight_channel_is_a_noop(self, monkeypatch): - runner._immediate_backfill.add("UC_busy") + runner._backfill_in_progress.add("UC_busy") monkeypatch.setattr(runner, "SessionLocal", self._boom_session) - runner.backfill_channel_now("UC_busy") # returns immediately; _boom_session never runs - assert runner._immediate_backfill == {"UC_busy"} # the in-flight entry is untouched + runner.backfill_channel_now("UC_busy", 1) # returns immediately; _boom_session never runs + assert runner._backfill_in_progress == {"UC_busy"} # the in-flight entry is untouched def test_over_the_concurrency_cap_falls_back_to_the_scheduler(self, monkeypatch): - for i in range(runner._MAX_CONCURRENT_IMMEDIATE): - runner._immediate_backfill.add(f"UC_full_{i}") + runner._immediate_running = runner._MAX_CONCURRENT_IMMEDIATE monkeypatch.setattr(runner, "SessionLocal", self._boom_session) - runner.backfill_channel_now("UC_new") # over the cap → no-op, scheduler will handle it - assert "UC_new" not in runner._immediate_backfill + runner.backfill_channel_now("UC_new", 1) # over the cap → no-op, scheduler will handle it + assert "UC_new" not in runner._backfill_in_progress def test_scheduler_deep_pass_skips_an_in_flight_channel(self, db, monkeypatch): # A channel the scheduler's query WOULD pick (deep-requested, recent-synced, not done)… ch = Channel( id="UC_inflight", title="In-flight", - recent_synced_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + recent_synced_at=_PUB, uploads_playlist_id="UU_inflight", backfill_done=False, ) @@ -95,8 +147,6 @@ class TestImmediateBackfillGuard: me = User(email="svc@example.com") db.add(me) db.flush() - from app.models import Subscription - db.add(Subscription(user_id=me.id, channel_id=ch.id, deep_requested=True)) db.commit() @@ -109,13 +159,13 @@ class TestImmediateBackfillGuard: ) monkeypatch.setattr(runner, "reconcile_full_history", lambda _db: 0) - # …is skipped by run_deep_backfill while an immediate backfill owns it (no cursor race). - runner._immediate_backfill.add(ch.id) + # …is skipped by run_deep_backfill while an interactive backfill owns it (no cursor race). + runner._backfill_in_progress.add(ch.id) runner.run_deep_backfill(db) assert paged == [], "the scheduler must not page a channel an immediate backfill is handling" - # …and IS paged once the immediate backfill releases it. - runner._immediate_backfill.discard(ch.id) + # …and IS paged once the interactive backfill releases it. + runner._backfill_in_progress.discard(ch.id) runner.run_deep_backfill(db) assert paged == [ch.id] diff --git a/frontend/src/components/ChannelPage.tsx b/frontend/src/components/ChannelPage.tsx index 74a023d8..ca420456 100644 --- a/frontend/src/components/ChannelPage.tsx +++ b/frontend/src/components/ChannelPage.tsx @@ -2,6 +2,7 @@ import { useCallback, useEffect, useRef, useState } from "react"; import { useTranslation } from "react-i18next"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { useLiveQuery } from "../lib/useLiveQuery"; +import { channelDetailPollMs } from "../lib/channelDetailPoll"; import { qk } from "../lib/queryKeys"; import { useNavigationActions } from "./NavigationProvider"; import { PriorityStepper } from "./PriorityStepper"; @@ -85,7 +86,7 @@ export default function ChannelPage({ // scheduler tops up any tail on its ~10-minute ticks — then stop polling once the back-catalogue is // in. Same queryKey as before, so mutation invalidations still hit. const detail = useLiveQuery(qk.channel(channelId), () => api.channelDetail(channelId), { - intervalMs: (d) => (d?.deep_requested && !d.backfill_done ? 4000 : false), + intervalMs: channelDetailPollMs, }); const ch = detail.data; useEffect(() => setBannerAttempt(0), [ch?.banner_url]); // fresh channel/banner → try again diff --git a/frontend/src/components/channelColumns.tsx b/frontend/src/components/channelColumns.tsx index efe4e9ae..0f38fbe4 100644 --- a/frontend/src/components/channelColumns.tsx +++ b/frontend/src/components/channelColumns.tsx @@ -52,6 +52,7 @@ function StoredCell({ c }: { c: ManagedChannel }) { {isFullySynced(c) && ( diff --git a/frontend/src/lib/channelDetailPoll.test.ts b/frontend/src/lib/channelDetailPoll.test.ts new file mode 100644 index 00000000..03c7a446 --- /dev/null +++ b/frontend/src/lib/channelDetailPoll.test.ts @@ -0,0 +1,24 @@ +import { describe, it, expect } from "vitest"; +import { channelDetailPollMs } from "./channelDetailPoll"; +import { type ChannelDetail } from "./api"; + +// Only deep_requested + backfill_done drive the decision; cast a partial for the pure predicate. +const d = (over: Partial): ChannelDetail => over as ChannelDetail; + +describe("channelDetailPollMs", () => { + it("polls at 4s while THIS user's full-history backfill is running", () => { + expect(channelDetailPollMs(d({ deep_requested: true, backfill_done: false }))).toBe(4000); + }); + + it("stops (false) once the back-catalogue is fully in", () => { + expect(channelDetailPollMs(d({ deep_requested: true, backfill_done: true }))).toBe(false); + }); + + it("stops (false) when the user hasn't requested full history", () => { + expect(channelDetailPollMs(d({ deep_requested: false, backfill_done: false }))).toBe(false); + }); + + it("stops (false) before the detail has loaded", () => { + expect(channelDetailPollMs(undefined)).toBe(false); + }); +}); diff --git a/frontend/src/lib/channelDetailPoll.ts b/frontend/src/lib/channelDetailPoll.ts new file mode 100644 index 00000000..0f796e51 --- /dev/null +++ b/frontend/src/lib/channelDetailPoll.ts @@ -0,0 +1,8 @@ +import { type ChannelDetail } from "./api"; + +/** How often the channel page re-polls its detail (via useLiveQuery): every 4s while THIS user's + * full-history backfill is running (`deep_requested && !backfill_done`) so the stored count + the + * N/S/L breakdown climb on their own, and `false` (stop polling) once the back-catalogue is in or + * the user never requested it. Extracted so the poll-or-stop decision is unit-testable. */ +export const channelDetailPollMs = (d: ChannelDetail | undefined): number | false => + d?.deep_requested && !d.backfill_done ? 4000 : false;