From 0517de1f59e91a5c794bf19272e69196557e2e11 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 17 Sep 2026 00:38:46 +0200 Subject: [PATCH] fix: poll and backfill only channels somebody follows The scheduled RSS poll and recent backfill covered every channel row, but rows also come from search results, playlist sync and dropped subscriptions, and nothing deletes them. On prod 582 of 979 polled channels had no follower, three of them dead feeds answering 404 on every poll. Both jobs now take subscribed or explored channels only, and the scheduler card counts the same set. As a deliberate trade-off, the shared Library no longer receives new uploads from channels nobody follows. --- backend/app/routes/scheduler.py | 8 +++- backend/app/sync/runner.py | 28 +++++++++-- backend/tests/test_followed_channels.py | 63 +++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 7 deletions(-) create mode 100644 backend/tests/test_followed_channels.py 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