peter
48285a715b
Addresses the code-review findings on the immediate full-history change: - Extract quota.backfill_reserve_reached(): the remaining/reserve compare was open-coded at seven sites (recent/deep/immediate backfill, explore, rolling revalidation, and the new per-page guard). It reads in its OWN short-lived session (like record_usage/log_action), so a per-page guard no longer holds the caller's transaction open across the YouTube fetch that follows — the 400-page immediate run no longer pins a pooled connection idle-in-transaction through each network round-trip. Migrated all call sites; dropped the now-dead sysconfig imports in channels.py and maintenance.py. - Simplify the deep-backfill loop: channel.backfill_done = not page_token (done is False on entry) plus one break, replacing the doubled `if not page_token`. - Tests patch the backfill_reserve_reached seam directly instead of mocking remaining_today internals, removing the fragile dead-value budget iterator. Full backend suite green; ruff clean.
338 lines
15 KiB
Python
338 lines
15 KiB
Python
"""Reusable sync routines shared by the scheduler and the manual /api/sync endpoints.
|
|
|
|
Channel/video data is shared across users, so background work acts through a "service
|
|
user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configured for
|
|
public reads.
|
|
"""
|
|
import logging
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import progress, quota, sysconfig
|
|
|
|
log = logging.getLogger("siftlode.sync")
|
|
from app.db import SessionLocal
|
|
from app.models import Channel, OAuthToken, Subscription, User
|
|
from app.sync.subscriptions import import_subscriptions
|
|
from app.sync.videos import (
|
|
apply_rss_feed,
|
|
backfill_channel_deep,
|
|
backfill_channel_recent,
|
|
enrich_pending,
|
|
reconcile_full_history,
|
|
refresh_live,
|
|
run_shorts_classification,
|
|
)
|
|
from app.youtube.client import YouTubeClient
|
|
from app.youtube.rss import RssError, fetch_channel_feed
|
|
|
|
# RSS feeds are free and network-bound, so fetch many at once; DB writes stay single-threaded.
|
|
RSS_POLL_WORKERS = 16
|
|
|
|
|
|
def token_users(db: Session) -> list[User]:
|
|
"""Every user with a stored YouTube refresh token (the read-scope service accounts), by id."""
|
|
return list(
|
|
db.execute(
|
|
select(User)
|
|
.join(OAuthToken)
|
|
.where(OAuthToken.refresh_token_enc.is_not(None))
|
|
.order_by(User.id)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def get_service_user(db: Session) -> User | None:
|
|
users = token_users(db)
|
|
return users[0] if users else None
|
|
|
|
|
|
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
|
|
_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).
|
|
|
|
`failed` counts a channel whose feed could not be READ *or* whose rows could not be APPLIED —
|
|
both leave the channel un-polled, and counting only the first would move the blind spot rather
|
|
than remove it.
|
|
|
|
Raises RuntimeError when NOTHING could be polled (`failed == total`, total > 0): a run that
|
|
achieved literally nothing is a failed run, and only an exception turns the scheduler card red
|
|
— a returned count leaves it green with the outage buried in the summary line. The message
|
|
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()
|
|
total = len(channels)
|
|
new = 0
|
|
done = 0
|
|
failed = 0
|
|
# Kept for the total-outage message: "failed for all N channels" alone can't tell an egress
|
|
# outage from one permanently dead feed id — which is the whole difference on an instance with
|
|
# a single subscription, where BOTH render as 100% failed.
|
|
first_error: str | None = None
|
|
# Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as
|
|
# each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a
|
|
# plain str) to the workers; the ORM instance is touched only here on the main thread.
|
|
futures = {}
|
|
with ThreadPoolExecutor(max_workers=RSS_POLL_WORKERS) as pool:
|
|
for channel in channels:
|
|
futures[pool.submit(fetch_channel_feed, channel.id)] = channel
|
|
for fut in as_completed(futures):
|
|
channel = futures[fut]
|
|
try:
|
|
entries = fut.result()
|
|
# A failed poll is NOT an empty feed: skip the apply so `last_rss_at` stays where it
|
|
# was (stamping it made a permanently-broken feed look freshly polled). The shared
|
|
# counter/progress tail lives after the statement, so the two outcomes can stay two
|
|
# clauses: an RssError is EXPECTED (404, timeout) and gets one warning line; anything
|
|
# else is a bug on our side and earns a traceback.
|
|
except RssError as exc:
|
|
log.warning("RSS poll skipped for channel %s: %s", channel.id, exc)
|
|
failed += 1
|
|
first_error = first_error or f"{channel.id}: {exc}"
|
|
except Exception as exc:
|
|
log.exception("RSS fetch failed for channel %s", channel.id)
|
|
failed += 1
|
|
first_error = first_error or f"{channel.id}: {exc}"
|
|
else:
|
|
try:
|
|
new += apply_rss_feed(db, channel, entries)
|
|
except Exception as exc:
|
|
# A write failure (read-only DB, failover) leaves the channel just as un-polled
|
|
# as an unreachable feed — count it, or a wholly broken run reports "failed=0".
|
|
db.rollback()
|
|
failed += 1
|
|
first_error = first_error or f"{channel.id} (apply): {exc}"
|
|
log.exception("RSS apply failed for channel %s", channel.id)
|
|
# One place that advances the counter, whatever happened above.
|
|
done += 1
|
|
progress.report(done, total, "rss_poll")
|
|
if failed:
|
|
log.warning("RSS poll: %d/%d channels could not be polled", failed, total)
|
|
if total and failed == total:
|
|
# Nothing was polled at all — an egress outage (the tinyproxy-after-reboot incident), a
|
|
# dead DNS, a read-only database. Returning normally would record the run as `ok` with the
|
|
# failure hidden inside the result string, i.e. a green pill for a total outage.
|
|
# The first error TRAVELS WITH the message on purpose: on an instance with one or two
|
|
# subscriptions "all channels failed" is also what ONE dead feed id looks like, and a red
|
|
# card the user can't act on is its own kind of dishonest. `channel: HTTP 404` says
|
|
# "fix this subscription"; `Connection refused` says "the egress is down".
|
|
raise RuntimeError(f"RSS poll failed for all {total} channels — first: {first_error}")
|
|
return {"new": new, "failed": failed}
|
|
|
|
|
|
def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
|
|
user = get_service_user(db)
|
|
if user is None:
|
|
return 0
|
|
total = 0
|
|
with YouTubeClient(db, user) as yt:
|
|
for _ in range(max_batches):
|
|
if quota.remaining_today(db) <= floor:
|
|
break
|
|
n = enrich_pending(db, yt)
|
|
total += n
|
|
progress.report(total, None, "enrich") # total unknown (drains until empty)
|
|
if n == 0:
|
|
break
|
|
# Revisit transient live/upcoming videos (enrich_pending is one-shot) so an ended
|
|
# stream gains its final duration + was_live status instead of staying "live" forever.
|
|
if quota.remaining_today(db) > floor:
|
|
total += refresh_live(db, yt)
|
|
return total
|
|
|
|
|
|
def run_shorts(db: Session) -> dict:
|
|
return run_shorts_classification(db)
|
|
|
|
|
|
def run_subscription_resync(db: Session) -> dict:
|
|
"""Re-import every user's subscriptions so unsubscribes and new subscriptions on
|
|
YouTube are reflected automatically."""
|
|
users = token_users(db)
|
|
for i, user in enumerate(users, 1):
|
|
try:
|
|
import_subscriptions(db, user)
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("Subscription resync failed for user %s", user.id)
|
|
progress.report(i, len(users), "subscriptions")
|
|
return {"users": len(users)}
|
|
|
|
|
|
def run_recent_backfill(
|
|
db: Session, channels: list[Channel] | None = None, max_channels: int | None = None
|
|
) -> dict:
|
|
user = get_service_user(db)
|
|
if user is None:
|
|
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()
|
|
)
|
|
if max_channels:
|
|
channels = channels[:max_channels]
|
|
processed = 0
|
|
videos_added = 0
|
|
with YouTubeClient(db, user) as yt:
|
|
for channel in channels:
|
|
if quota.backfill_reserve_reached():
|
|
break
|
|
try:
|
|
videos_added += backfill_channel_recent(db, yt, channel)
|
|
processed += 1
|
|
progress.report(processed, len(channels), "backfill_recent")
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("Recent backfill failed for channel %s", channel.id)
|
|
reconciled = reconcile_full_history(db)
|
|
return {
|
|
"channels_processed": processed,
|
|
"videos_added": videos_added,
|
|
"reconciled_full": reconciled,
|
|
}
|
|
|
|
|
|
def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -> dict:
|
|
user = get_service_user(db)
|
|
if user is None:
|
|
return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
|
|
# Demand-driven: only deep-backfill channels at least one user has opted into
|
|
# (Subscription.deep_requested). Recent backfill stays unconditional (cheap).
|
|
deep_requested = (
|
|
select(Subscription.id)
|
|
.where(
|
|
Subscription.channel_id == Channel.id,
|
|
Subscription.deep_requested.is_(True),
|
|
)
|
|
.exists()
|
|
)
|
|
channels = (
|
|
db.execute(
|
|
select(Channel).where(
|
|
Channel.backfill_done.is_(False),
|
|
Channel.recent_synced_at.is_not(None),
|
|
deep_requested,
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)[:max_channels]
|
|
processed = 0
|
|
videos_added = 0
|
|
with YouTubeClient(db, user) as yt:
|
|
for channel in channels:
|
|
if quota.backfill_reserve_reached():
|
|
break
|
|
# Claim the channel (or skip it if an interactive backfill already owns it) so the two
|
|
# never advance the same cursor concurrently — symmetric with backfill_channel_now.
|
|
with _backfill_lock:
|
|
if channel.id in _backfill_in_progress:
|
|
continue
|
|
_backfill_in_progress.add(channel.id)
|
|
try:
|
|
videos_added += backfill_channel_deep(db, yt, channel, max_pages)
|
|
processed += 1
|
|
progress.report(processed, len(channels), "backfill_deep")
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("Deep backfill failed for channel %s", channel.id)
|
|
finally:
|
|
with _backfill_lock:
|
|
_backfill_in_progress.discard(channel.id)
|
|
reconciled = reconcile_full_history(db)
|
|
return {
|
|
"channels_processed": processed,
|
|
"videos_added": videos_added,
|
|
"reconciled_full": reconciled,
|
|
}
|
|
|
|
|
|
# Channels being deep-backfilled RIGHT NOW by EITHER path — the interactive backfill_channel_now (a
|
|
# FastAPI BackgroundTasks threadpool thread) OR the scheduler's run_deep_backfill (its own thread).
|
|
# BOTH register the channel they're paging (under the lock) so the other skips it: no two runs page
|
|
# the same channel's `backfill_cursor` concurrently, whichever started first. `_immediate_running`
|
|
# counts only the interactive path, capped so one user's burst can't drain the shared budget (the
|
|
# scheduler is already quota-paced). Single uvicorn process → one shared module state, lock suffices.
|
|
_backfill_in_progress: set[str] = set()
|
|
_immediate_running = 0
|
|
_backfill_lock = threading.Lock()
|
|
_MAX_CONCURRENT_IMMEDIATE = 2
|
|
|
|
|
|
def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 400) -> 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). max_pages is high enough to finish
|
|
any realistic channel in a single run (400 pages = 20 000 videos; e.g. BBC Earth's ~3400 = 69
|
|
pages — the old 60-page cap stopped one page short of large channels, so their tail dribbled in
|
|
over the scheduler's 10-minute ticks and looked stuck). backfill_channel_deep re-checks the shared
|
|
budget before each page and yields at the reserve, so a high cap still can't drain the day's quota.
|
|
Quota-guarded, capped to _MAX_CONCURRENT_IMMEDIATE interactive runs so one user can't drain the
|
|
shared backfill budget by opting many channels in at once (the overflow, and any channel too large
|
|
to finish under the reserve, is finished by the scheduler since deep_requested stays set), and
|
|
mutually excluded per-channel with the scheduler via _backfill_in_progress. Best-effort — logs and
|
|
swallows failures."""
|
|
global _immediate_running
|
|
with _backfill_lock:
|
|
if channel_id in _backfill_in_progress or _immediate_running >= _MAX_CONCURRENT_IMMEDIATE:
|
|
return # already backfilling this channel, or too many interactive runs — leave it to the scheduler
|
|
_backfill_in_progress.add(channel_id)
|
|
_immediate_running += 1
|
|
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.backfill_reserve_reached():
|
|
return
|
|
try:
|
|
# Attribute the spend to the requesting user under the full-backfill action, so it
|
|
# shows up correctly in the usage breakdown (not the default OTHER/system bucket).
|
|
with quota.attribute(user_id, quota.QuotaAction.VIDEOS_BACKFILL_FULL):
|
|
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 _backfill_lock:
|
|
_backfill_in_progress.discard(channel_id)
|
|
_immediate_running -= 1
|
|
|
|
|
|
def estimate_deep_backfill(db: Session, channels: list[Channel]) -> dict:
|
|
"""Rough ETA to finish full-history backfill of `channels`. Quota-bound: total
|
|
estimated API units / the daily budget left for backfill. Deliberately approximate —
|
|
a channel's cost ~= playlistItems pages (1 unit / 50 videos) + enrichment (likewise)."""
|
|
pending = [c for c in channels if not c.backfill_done]
|
|
|
|
def units(c: Channel) -> int:
|
|
videos = c.video_count or 0
|
|
pages = max(1, (videos + 49) // 50)
|
|
return 2 * pages
|
|
|
|
total_units = sum(units(c) for c in pending)
|
|
daily_budget = max(
|
|
1, sysconfig.get_int(db, "quota_daily_budget") - sysconfig.get_int(db, "backfill_quota_reserve")
|
|
)
|
|
eta_seconds = int(total_units / daily_budget * 86_400) if pending else 0
|
|
return {
|
|
"channels_pending": len(pending),
|
|
"videos_pending_est": sum(c.video_count or 0 for c in pending),
|
|
"units_est": total_units,
|
|
"daily_budget": daily_budget,
|
|
"eta_seconds": eta_seconds,
|
|
}
|