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.
This commit is contained in:
@@ -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"],
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user