diff --git a/backend/app/routes/scheduler.py b/backend/app/routes/scheduler.py index c5a39258..60489bbe 100644 --- a/backend/app/routes/scheduler.py +++ b/backend/app/routes/scheduler.py @@ -22,7 +22,7 @@ from app.scheduler import ( trigger_all, trigger_job, ) -from app.sync.runner import estimate_deep_backfill +from app.sync.runner import estimate_deep_backfill, followed_channel from app.routes._common import not_found router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"]) @@ -137,7 +137,11 @@ def get_scheduler(_: User = Depends(admin_user), db: Session = Depends(get_db)) queue = { "channels_recent_pending": db.scalar( - select(func.count()).select_from(Channel).where(Channel.recent_synced_at.is_(None)) + # The same selection run_recent_backfill works through, or the card counts channels + # the job will never touch and never reaches zero. + select(func.count()) + .select_from(Channel) + .where(Channel.recent_synced_at.is_(None), followed_channel()) ) or 0, "channels_deep_pending": deep_eta["channels_pending"], diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index e64af62e..24111d77 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -10,14 +10,14 @@ import threading from collections import Counter from concurrent.futures import ThreadPoolExecutor, as_completed -from sqlalchemy import select +from sqlalchemy import or_, select from sqlalchemy.orm import Session from app import logs, progress, quota, sysconfig log = logging.getLogger("siftlode.sync") from app.db import SessionLocal -from app.models import Channel, OAuthToken, Subscription, User +from app.models import Channel, ExploredChannel, OAuthToken, Subscription, User from app.sync.subscriptions import import_subscriptions from app.sync.videos import ( apply_rss_feed, @@ -54,8 +54,22 @@ def get_service_user(db: Session) -> User | None: return users[0] if users else None +def followed_channel(): + """WHERE clause: a channel somebody follows — a subscription, or an exploration still inside its + grace window. The scheduled RSS poll and recent backfill cover only these (owner decision, + 2026-09-17). They used to cover EVERY channel row, but rows also come from search results, + playlist sync and dropped subscriptions, and nothing deletes those: on prod 582 of 979 polled + channels had no follower, three of them dead feeds answering 404 on every poll. The cost of + narrowing is deliberate: the shared Library stops receiving new uploads from channels nobody + follows.""" + return or_( + select(Subscription.id).where(Subscription.channel_id == Channel.id).exists(), + select(ExploredChannel.id).where(ExploredChannel.channel_id == Channel.id).exists(), + ) + + def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str, int]: - """Poll every channel's free RSS feed. Returns `{"new": inserted, "failed": channels_that_did + """Poll the free RSS feed of every followed channel (`followed_channel`). Returns `{"new": inserted, "failed": channels_that_did _not_complete}` — the failure count is part of the RESULT, not just a log line, so the scheduler card and the audit summary can tell "no new uploads" apart from "nothing worked" (an egress outage, or a database that can't accept writes, otherwise renders identically to a quiet hour). @@ -70,7 +84,7 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str carries the FIRST underlying error, because on a small instance "all channels failed" is also what a single dead feed id looks like.""" if channels is None: - channels = db.execute(select(Channel)).scalars().all() + channels = db.execute(select(Channel).where(followed_channel())).scalars().all() total = len(channels) new = 0 done = 0 @@ -210,7 +224,11 @@ def run_recent_backfill( return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"} if channels is None: channels = ( - db.execute(select(Channel).where(Channel.recent_synced_at.is_(None))).scalars().all() + db.execute( + select(Channel).where(Channel.recent_synced_at.is_(None), followed_channel()) + ) + .scalars() + .all() ) if max_channels: channels = channels[:max_channels] diff --git a/backend/tests/test_followed_channels.py b/backend/tests/test_followed_channels.py new file mode 100644 index 00000000..798b972f --- /dev/null +++ b/backend/tests/test_followed_channels.py @@ -0,0 +1,63 @@ +"""The scheduled RSS poll and recent backfill cover FOLLOWED channels only (owner decision, 2026-09-17). + +A channel row also comes from a search result, a playlist item or a subscription somebody dropped, +and nothing deletes those rows. Polling all of them was 582 of 979 channels on prod with no follower, +three of them dead feeds answering 404 every 20 minutes. Followed = subscribed by someone, or explored +by someone. The admin card counting "channels to sync" must use the same selection, or it counts +channels the backfill will never touch. DB lane. +""" + +from app.models import Channel, ExploredChannel, Subscription +from app.sync import runner + + +def _channels(db, make_user): + user = make_user() + for cid in ("UC_sub", "UC_explored", "UC_search", "UC_playlist"): + db.add(Channel(id=cid, title=cid, from_search=cid == "UC_search")) + db.flush() + db.add(Subscription(user_id=user.id, channel_id="UC_sub")) + db.add(ExploredChannel(user_id=user.id, channel_id="UC_explored")) + db.commit() + return user + + +def test_the_rss_poll_skips_channels_nobody_follows(db, make_user, monkeypatch): + _channels(db, make_user) + polled: list[str] = [] + monkeypatch.setattr(runner, "fetch_channel_feed", lambda cid: polled.append(cid) or []) + monkeypatch.setattr(runner, "apply_rss_feed", lambda db, ch, entries: 0) + assert runner.run_rss_poll(db) == {"new": 0, "failed": 0} + assert sorted(polled) == ["UC_explored", "UC_sub"] + + +def test_the_recent_backfill_skips_channels_nobody_follows(db, make_user, monkeypatch): + user = _channels(db, make_user) + done: list[str] = [] + + class _Client: + def __init__(self, *a): + pass + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + monkeypatch.setattr(runner, "get_service_user", lambda db: user) + monkeypatch.setattr(runner, "YouTubeClient", _Client) + monkeypatch.setattr(runner.quota, "backfill_reserve_reached", lambda: False) + monkeypatch.setattr( + runner, "backfill_channel_recent", lambda db, yt, ch: done.append(ch.id) or 0 + ) + monkeypatch.setattr(runner, "reconcile_full_history", lambda db: 0) + runner.run_recent_backfill(db) + assert sorted(done) == ["UC_explored", "UC_sub"] + + +def test_the_dashboard_counts_the_channels_the_backfill_will_process(db, make_user, as_user): + _channels(db, make_user) + admin = make_user(role="admin") + queue = as_user(admin).get("/api/admin/scheduler").json()["queue"] + assert queue["channels_recent_pending"] == 2