peter
b59fd38a9c
The log now has a stored, queryable copy for the diagnostics console: every WARNING and above plus the request line of every request, kept for log_retention_days (default 7, no "forever") and pruned hourly by the new log_gc job. Rows are built on the logging thread from the same redacted fields as the stdout line, queued without blocking and written in batches by one background thread on its own connection. A full queue or a failed write drops and reports instead of retrying. Flood control keeps 30 of any one message per minute plus a count of the rest; anonymous request lines are capped at 60 a minute per client, and the client address is never stored. Deleting an account deletes its stored events. Measured on prod first: a YouTube RSS outage wrote 11k warnings in four hours, one per channel per poll. The poll now writes one warning with per-reason counts and a few examples, and a failure on our side gets one traceback per poll instead of one per channel.
375 lines
17 KiB
Python
375 lines
17 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 collections import Counter
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
from sqlalchemy import 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.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
|
|
# Failed channels named in a poll's summary warning; the rest are counted by reason.
|
|
RSS_FAILURE_EXAMPLES = 5
|
|
|
|
|
|
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
|
|
# What failed, as counts per reason (`http_404`, `transport`, `apply`, …) plus a few examples —
|
|
# the ONE warning a poll writes. One line per failed channel was what turned YouTube's RSS outage
|
|
# of 2026-09-16 into 11k warnings in four hours; each channel's own line is DEBUG now.
|
|
reasons: Counter[str] = Counter()
|
|
examples: list[str] = []
|
|
|
|
def fail(reason: str, detail: str, *, unexpected: str | None = None) -> None:
|
|
"""Count a channel that did not get polled. `unexpected` is the message for a failure that
|
|
is a bug on our side: its FIRST occurrence per reason gets a traceback at ERROR, the rest
|
|
DEBUG — a read-only database fails every channel the same way, and 979 identical tracebacks
|
|
tell nobody anything the first one didn't. Call it from inside the `except` block."""
|
|
nonlocal failed, first_error
|
|
if unexpected is not None:
|
|
level = logging.ERROR if reasons[reason] == 0 else logging.DEBUG
|
|
log.log(level, unexpected, exc_info=True)
|
|
failed += 1
|
|
reasons[reason] += 1
|
|
first_error = first_error or detail
|
|
if len(examples) < RSS_FAILURE_EXAMPLES:
|
|
examples.append(detail)
|
|
|
|
# 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 is only counted; anything else is a
|
|
# bug on our side and earns a traceback (the first of its kind per poll).
|
|
except RssError as exc:
|
|
log.debug("RSS poll skipped for channel %s: %s", channel.id, exc)
|
|
fail(exc.reason, f"{channel.id}: {exc}")
|
|
except Exception as exc:
|
|
fail(
|
|
"error", f"{channel.id}: {exc}", unexpected=f"RSS fetch failed for {channel.id}"
|
|
)
|
|
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()
|
|
fail(
|
|
"apply",
|
|
f"{channel.id} (apply): {exc}",
|
|
unexpected=f"RSS apply failed for {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,
|
|
extra=logs.event(
|
|
"sync.rss.failed",
|
|
failed=failed,
|
|
total=total,
|
|
reasons=dict(reasons),
|
|
examples=examples,
|
|
),
|
|
)
|
|
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,
|
|
}
|