fix(channels): address R14 code-review findings
- 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).
This commit is contained in:
@@ -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,
|
||||
|
||||
+41
-19
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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),
|
||||
});
|
||||
|
||||
@@ -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) =>
|
||||
|
||||
@@ -19,9 +19,6 @@ const ch = (over: Partial<ManagedChannel>): 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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user