From 2cc77757d9ee5f459aaacdecfc2b0b37d3c04374 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 15 Aug 2026 06:38:19 +0200 Subject: [PATCH] fix(channels): address R14 code-review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - list_channels no longer computes the Normal/Shorts/Live aggregation the frontend stopped reading when R14 dropped the N/S/L table column; the dead ManagedChannel.count_* fields are gone too. - Immediate "get full history" backfill now coordinates with the paced scheduler via a small in-flight set: capped at 2 concurrent + per-channel dedup, and run_deep_backfill skips a channel an immediate backfill owns — killing the cursor race and bounding the quota-drain burst, without losing the speed win. - Add test_channel_backfill.py: the channelDetail N/S/L counts + the guard's no-op / cap / scheduler-skip behaviour. - Inline the syncedChannels useMemo (its dep is a fresh array each render, so it never cached); fix the stale "~2-min ticks" comment (it's 10 minutes). --- backend/app/routes/channels.py | 15 +- backend/app/sync/runner.py | 60 +++++--- backend/tests/test_channel_backfill.py | 128 ++++++++++++++++++ frontend/src/components/ChannelPage.tsx | 5 +- frontend/src/components/Channels.tsx | 7 +- .../components/channelColumns.dom.test.tsx | 3 - frontend/src/lib/api/channels.ts | 3 - 7 files changed, 179 insertions(+), 42 deletions(-) create mode 100644 backend/tests/test_channel_backfill.py diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py index 462c69f9..2a29c6ec 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -50,20 +50,19 @@ def list_channels( ).all() # Per-channel aggregates over the shared catalog in ONE grouped pass: stored count, last - # upload, total duration, and a Normal/Shorts/Live breakdown. Cheap — a single grouped scan - # (~35 ms over the full 233k-row catalog on the dev DB), so computed on read rather than + # upload, total duration. Cheap — a single grouped scan, so computed on read rather than # denormalised (measure-first: reads aren't frequent/hot enough to warrant cached columns). + # (The Normal/Shorts/Live breakdown moved to the channel-detail page in R14, so it is no longer + # computed here — the manager table dropped its N/S/L column.) channel_ids = [c.id for _, c in rows] agg: dict[str, dict] = {} if channel_ids: - for cid, n, last_up, total_dur, shorts, live in db.execute( + for cid, n, last_up, total_dur in db.execute( select( Video.channel_id, func.count(Video.id), func.max(Video.published_at), func.coalesce(func.sum(Video.duration_seconds), 0), - func.count(Video.id).filter(Video.is_short.is_(True)), - func.count(Video.id).filter(Video.live_status.in_(LIVE_OR_UPCOMING)), ) .where(Video.channel_id.in_(channel_ids)) .group_by(Video.channel_id) @@ -72,9 +71,6 @@ def list_channels( "stored": n, "last_video_at": iso(last_up), "total_duration_seconds": int(total_dur or 0), - "count_short": int(shorts), - "count_live": int(live), - "count_normal": max(0, int(n) - int(shorts) - int(live)), } # Channels already in the *global* deep-backfill queue — i.e. at least one user (any @@ -108,9 +104,6 @@ def list_channels( "stored_videos": (agg.get(ch.id) or {}).get("stored", 0), "last_video_at": (agg.get(ch.id) or {}).get("last_video_at"), "total_duration_seconds": (agg.get(ch.id) or {}).get("total_duration_seconds", 0), - "count_normal": (agg.get(ch.id) or {}).get("count_normal", 0), - "count_short": (agg.get(ch.id) or {}).get("count_short", 0), - "count_live": (agg.get(ch.id) or {}).get("count_live", 0), "priority": sub.priority, "hidden": sub.hidden, "deep_requested": sub.deep_requested, diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index ae7fe21c..4849daf9 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -5,6 +5,7 @@ user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configu public reads. """ import logging +import threading from concurrent.futures import ThreadPoolExecutor, as_completed from sqlalchemy import select @@ -230,6 +231,8 @@ 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 try: videos_added += backfill_channel_deep(db, yt, channel, max_pages) processed += 1 @@ -245,28 +248,47 @@ 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() +_MAX_CONCURRENT_IMMEDIATE = 2 + + def backfill_channel_now(channel_id: str, 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; a very large - channel's tail (beyond ~max_pages*50 videos) is finished by the scheduler on its normal runs, - since deep_requested stays set. Best-effort — logs and swallows failures.""" - with SessionLocal() as db: - user = get_service_user(db) - if user is None: - return - 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"): - return - try: - 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) + 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) + try: + with SessionLocal() as db: + user = get_service_user(db) + if user is None: + return + 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"): + return + try: + 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) 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 new file mode 100644 index 00000000..a70a3296 --- /dev/null +++ b/backend/tests/test_channel_backfill.py @@ -0,0 +1,128 @@ +"""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.""" +from datetime import datetime, timezone + +from fastapi.testclient import TestClient + +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.sync import runner + + +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), + ) + 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. + 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"), + ] + ) + db.commit() + + # The request gate 503s an unconfigured instance; flip the process-level flag (auto-restored). + monkeypatch.setattr(state, "_configured_cache", True) + app.dependency_overrides[get_db] = lambda: db + app.dependency_overrides[current_user] = lambda: me + try: + resp = TestClient(app).get(f"/api/channels/{ch.id}") + finally: + app.dependency_overrides.clear() + + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["stored_videos"] == 4 + assert body["count_short"] == 1 + assert body["count_live"] == 1 + assert body["count_normal"] == 2 + + +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.""" + + def setup_method(self): + runner._immediate_backfill.clear() + + def teardown_method(self): + runner._immediate_backfill.clear() + + @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. + 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") + 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 + + 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}") + 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 + + 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), + uploads_playlist_id="UU_inflight", + backfill_done=False, + ) + db.add(ch) + 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() + + paged: list[str] = [] + monkeypatch.setattr(runner, "get_service_user", lambda _db: me) + monkeypatch.setattr(runner.quota, "remaining_today", lambda _db: 10_000) + monkeypatch.setattr(runner, "YouTubeClient", lambda *a, **k: _NullYt()) + monkeypatch.setattr( + runner, "backfill_channel_deep", lambda _db, _yt, channel, _mp: paged.append(channel.id) or 0 + ) + 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) + 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) + runner.run_deep_backfill(db) + assert paged == [ch.id] + + +class _NullYt: + def __enter__(self): + return self + + def __exit__(self, *a): + return False diff --git a/frontend/src/components/ChannelPage.tsx b/frontend/src/components/ChannelPage.tsx index 4241c3c4..74a023d8 100644 --- a/frontend/src/components/ChannelPage.tsx +++ b/frontend/src/components/ChannelPage.tsx @@ -81,8 +81,9 @@ export default function ChannelPage({ const [bannerAttempt, setBannerAttempt] = useState(0); // A LIVE query: while THIS user's full-history backfill is running, poll so the stored count (and - // the N/S/L breakdown) climb on their own — the scheduler adds videos in ~2-min ticks — then stop - // polling once the back-catalogue is in. Same queryKey as before, so mutation invalidations still hit. + // the N/S/L breakdown) climb on their own — the immediate backfill fills it in seconds, and the + // 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), }); diff --git a/frontend/src/components/Channels.tsx b/frontend/src/components/Channels.tsx index 45cf51c5..366e6e71 100644 --- a/frontend/src/components/Channels.tsx +++ b/frontend/src/components/Channels.tsx @@ -249,10 +249,9 @@ export default function Channels() { (c) => !q || `${c.title ?? ""} ${c.handle ?? ""}`.toLowerCase().includes(q), ); // Sync-status pre-filter (top chips → `filters.sync`), applied before the table/cards see the rows. - const syncedChannels = useMemo( - () => applySyncFilter(visibleChannels, filters.sync), - [visibleChannels, filters.sync], - ); + // (Inline, not memoised: `visibleChannels` is a fresh array each render, so a useMemo keyed on it + // could never hit its cache — and applySyncFilter is a cheap single pass anyway.) + const syncedChannels = applySyncFilter(visibleChannels, filters.sync); const s = statusQuery.data; const onView = (c: ManagedChannel) => diff --git a/frontend/src/components/channelColumns.dom.test.tsx b/frontend/src/components/channelColumns.dom.test.tsx index f9272f59..a78ec3bb 100644 --- a/frontend/src/components/channelColumns.dom.test.tsx +++ b/frontend/src/components/channelColumns.dom.test.tsx @@ -19,9 +19,6 @@ const ch = (over: Partial): ManagedChannel => ({ stored_videos: 0, last_video_at: null, total_duration_seconds: 0, - count_normal: 0, - count_short: 0, - count_live: 0, priority: 0, hidden: false, deep_requested: false, diff --git a/frontend/src/lib/api/channels.ts b/frontend/src/lib/api/channels.ts index f9cd4605..c5113e54 100644 --- a/frontend/src/lib/api/channels.ts +++ b/frontend/src/lib/api/channels.ts @@ -73,9 +73,6 @@ export interface ManagedChannel { stored_videos: number; last_video_at: string | null; total_duration_seconds: number; - count_normal: number; - count_short: number; - count_live: number; priority: number; hidden: boolean; deep_requested: boolean;