From 6eb552fa04206427f3e4725af35f95818c013984 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 19 Jun 2026 02:43:37 +0200 Subject: [PATCH 1/2] feat(scheduler): report progress from every job + parallelize RSS polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every background job now reports progress, so the dashboard can show a live bar for any run (not only enrich/backfill/shorts/maintenance): rss_poll and subscriptions (runner), autotag, and playlist_sync gained progress.report calls over their loops. RSS polling now fetches the channel feeds concurrently (16 workers) instead of one-at-a-time — a slow/unreachable feed no longer blocks the rest, cutting a full poll of the catalogue from minutes to seconds. Feed fetches are network-bound and run in the pool; DB inserts stay on the session's thread (SQLAlchemy sessions aren't thread-safe), applied as each fetch completes. Split the DB-write half of poll_rss_channel into apply_rss_feed so the fetch and apply phases compose cleanly. --- backend/app/sync/autotag.py | 4 +++- backend/app/sync/playlists.py | 6 ++++-- backend/app/sync/runner.py | 39 ++++++++++++++++++++++++++++------- backend/app/sync/videos.py | 10 +++++++-- 4 files changed, 46 insertions(+), 13 deletions(-) diff --git a/backend/app/sync/autotag.py b/backend/app/sync/autotag.py index e10db9e..09cce68 100644 --- a/backend/app/sync/autotag.py +++ b/backend/app/sync/autotag.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Session log = logging.getLogger("subfeed.autotag") +from app import progress from app.config import settings from app.models import Channel, ChannelTag, Tag, Video @@ -270,13 +271,14 @@ def run_autotag_all(db: Session, only_missing: bool = False) -> dict: ) channels = db.execute(query).scalars().all() tagged = 0 - for channel in channels: + for i, channel in enumerate(channels, 1): try: apply_channel_autotags(db, channel) tagged += 1 except Exception: db.rollback() log.exception("Auto-tagging failed for channel %s", channel.id) + progress.report(i, len(channels), "autotag") removed = _cleanup_orphan_system_tags(db) return {"channels_tagged": tagged, "orphan_tags_removed": removed} diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py index bb60489..f21ef4f 100644 --- a/backend/app/sync/playlists.py +++ b/backend/app/sync/playlists.py @@ -12,7 +12,7 @@ from datetime import datetime, timezone from sqlalchemy import delete, select from sqlalchemy.orm import Session -from app import quota +from app import progress, quota from app.auth import has_read_scope, has_write_scope from app.models import Channel, OAuthToken, Playlist, PlaylistItem, User, Video from app.sync.videos import apply_video_details, parse_dt @@ -382,8 +382,9 @@ def sync_all_playlists(db: Session) -> dict: .all() ) total = 0 - for user in users: + for i, user in enumerate(users, 1): if not has_read_scope(user): + progress.report(i, len(users), "playlist_sync") continue try: with quota.attribute(user.id, "playlist_sync"): @@ -394,4 +395,5 @@ def sync_all_playlists(db: Session) -> dict: except Exception: db.rollback() log.exception("Playlist sync crashed for user %s", user.id) + progress.report(i, len(users), "playlist_sync") return {"users": len(users), "playlists_synced": total} diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 2a1a239..ed76ebc 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -5,6 +5,7 @@ user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configu public reads. """ import logging +from concurrent.futures import ThreadPoolExecutor, as_completed from sqlalchemy import select from sqlalchemy.orm import Session @@ -16,15 +17,19 @@ log = logging.getLogger("subfeed.sync") 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, - poll_rss_channel, reconcile_full_history, refresh_live, run_shorts_classification, ) from app.youtube.client import YouTubeClient +from app.youtube.rss import 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 get_service_user(db: Session) -> User | None: @@ -43,13 +48,30 @@ def get_service_user(db: Session) -> User | None: def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int: if channels is None: channels = db.execute(select(Channel)).scalars().all() + total = len(channels) new = 0 - for channel in channels: - try: - new += poll_rss_channel(db, channel) - except Exception: - db.rollback() - log.exception("RSS poll failed for channel %s", channel.id) + done = 0 + # 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() + except Exception: + log.exception("RSS fetch failed for channel %s", channel.id) + entries = [] + try: + new += apply_rss_feed(db, channel, entries) + except Exception: + db.rollback() + log.exception("RSS apply failed for channel %s", channel.id) + done += 1 + progress.report(done, total, "rss_poll") return new @@ -90,12 +112,13 @@ def run_subscription_resync(db: Session) -> dict: .scalars() .all() ) - for user in users: + 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)} diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index 0292390..4b64c51 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -67,8 +67,10 @@ def _insert_stubs(db: Session, rows: list[dict]) -> int: # --- RSS (free) --- -def poll_rss_channel(db: Session, channel: Channel) -> int: - entries = fetch_channel_feed(channel.id) +def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int: + """Insert new video stubs from an already-fetched feed and stamp the channel's last RSS + time. DB-only — must run on the session's own thread; the network fetch + (fetch_channel_feed) is kept separate so callers like run_rss_poll can parallelise it.""" inserted = _insert_stubs(db, entries) channel.last_rss_at = _now() db.add(channel) @@ -76,6 +78,10 @@ def poll_rss_channel(db: Session, channel: Channel) -> int: return inserted +def poll_rss_channel(db: Session, channel: Channel) -> int: + return apply_rss_feed(db, channel, fetch_channel_feed(channel.id)) + + # --- Backfill (uploads playlist; costs quota) --- def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None: content = item.get("contentDetails", {}) From cd026aaf79cf9adcc31341c67939e47d8e8e0897 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 19 Jun 2026 02:43:46 +0200 Subject: [PATCH 2/2] fix(scheduler): show progress for any running job with reliable live updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dashboard now renders a progress bar for any running job — determinate when counts are reported, an indeterminate "working" sliver otherwise — so a scheduled run is as visible as a manual one (progress was never manual-only; the wiring is shared, but only some jobs reported and the display gated on counts). Poll faster (1.5s) while any job runs, easing back to 4s when idle, derived from the freshest data by react-query's functional refetchInterval. The earlier React-state approach to this stalled the live updates (the row froze on the first sampled value); useLiveQuery now accepts a function of the data. Trilingual phase labels for the newly-reporting jobs. --- frontend/src/components/Scheduler.tsx | 24 +++++++++++++++------ frontend/src/i18n/locales/de/scheduler.json | 7 +++++- frontend/src/i18n/locales/en/scheduler.json | 7 +++++- frontend/src/i18n/locales/hu/scheduler.json | 7 +++++- frontend/src/lib/useLiveQuery.ts | 12 +++++++++-- 5 files changed, 45 insertions(+), 12 deletions(-) diff --git a/frontend/src/components/Scheduler.tsx b/frontend/src/components/Scheduler.tsx index 549916e..5a2556d 100644 --- a/frontend/src/components/Scheduler.tsx +++ b/frontend/src/components/Scheduler.tsx @@ -21,6 +21,8 @@ import { notify } from "../lib/notifications"; import Tooltip from "./Tooltip"; const POLL_MS = 4000; +// While any job is running, poll faster so short scheduled runs aren't missed between ticks. +const FAST_POLL_MS = 1500; // Seconds until an ISO instant (negative = past). function secsUntil(iso: string | null): number | null { @@ -73,19 +75,25 @@ function StatusLegend() { ); } -function JobProgress({ p }: { p: NonNullable }) { +// Shown for ANY running job. With reported counts it's a determinate bar; for a job that +// runs but doesn't report progress (or hasn't yet) it falls back to a generic "working" +// label and an indeterminate sliver — so every active run is visible, not just the ones +// that report numbers. +function JobProgress({ p }: { p: SchedulerJob["progress"] }) { const { t } = useTranslation(); - const phase = p.phase ? t(`scheduler.phase.${p.phase}`, p.phase) : null; + const phase = p?.phase ? t(`scheduler.phase.${p.phase}`, p.phase) : t("scheduler.phase.working"); const pct = - p.total && p.total > 0 ? Math.min(100, Math.round((p.current / p.total) * 100)) : null; + p?.total && p.total > 0 ? Math.min(100, Math.round((p.current / p.total) * 100)) : null; return (
{phase} - {p.total != null + {p?.total != null ? `${p.current.toLocaleString()} / ${p.total.toLocaleString()}` - : p.current.toLocaleString()} + : p + ? p.current.toLocaleString() + : ""}
@@ -184,7 +192,7 @@ function JobRow({ · {job.last_result} ) : null}
- {job.running && job.progress && } + {job.running && }
{job.running ? ( @@ -298,8 +306,10 @@ function Stat({ export default function Scheduler() { const { t } = useTranslation(); const qc = useQueryClient(); + // Poll faster while any job is running so live progress is smooth, then ease back when + // idle. Derived from the freshest data by react-query itself (see useLiveQuery). const q = useLiveQuery(["scheduler"], api.schedulerStatus, { - intervalMs: POLL_MS, + intervalMs: (d) => (d?.jobs?.some((j) => j.running) ? FAST_POLL_MS : POLL_MS), }); const data = q.data; diff --git a/frontend/src/i18n/locales/de/scheduler.json b/frontend/src/i18n/locales/de/scheduler.json index 8c30032..d1a451f 100644 --- a/frontend/src/i18n/locales/de/scheduler.json +++ b/frontend/src/i18n/locales/de/scheduler.json @@ -38,7 +38,12 @@ "enrich": "Videos anreichern", "backfill_recent": "Neue Uploads nachladen", "backfill_deep": "Ganze Historie nachladen", - "probe": "Shorts klassifizieren" + "probe": "Shorts klassifizieren", + "rss_poll": "Auf neue Uploads prüfen", + "subscriptions": "Abos synchronisieren", + "autotag": "Kanäle automatisch taggen", + "playlist_sync": "Playlists synchronisieren", + "working": "Arbeitet…" }, "maintenance": { "title": "Wartung", diff --git a/frontend/src/i18n/locales/en/scheduler.json b/frontend/src/i18n/locales/en/scheduler.json index b33bcf5..f9b0073 100644 --- a/frontend/src/i18n/locales/en/scheduler.json +++ b/frontend/src/i18n/locales/en/scheduler.json @@ -38,7 +38,12 @@ "enrich": "Enriching videos", "backfill_recent": "Backfilling recent uploads", "backfill_deep": "Backfilling full history", - "probe": "Classifying Shorts" + "probe": "Classifying Shorts", + "rss_poll": "Checking for new uploads", + "subscriptions": "Syncing subscriptions", + "autotag": "Auto-tagging channels", + "playlist_sync": "Syncing playlists", + "working": "Working…" }, "maintenance": { "title": "Maintenance", diff --git a/frontend/src/i18n/locales/hu/scheduler.json b/frontend/src/i18n/locales/hu/scheduler.json index a3a1b9a..7a8bc36 100644 --- a/frontend/src/i18n/locales/hu/scheduler.json +++ b/frontend/src/i18n/locales/hu/scheduler.json @@ -38,7 +38,12 @@ "enrich": "Videók dúsítása", "backfill_recent": "Friss feltöltések behúzása", "backfill_deep": "Teljes előzmény behúzása", - "probe": "Shorts besorolás" + "probe": "Shorts besorolás", + "rss_poll": "Új feltöltések keresése", + "subscriptions": "Feliratkozások szinkronizálása", + "autotag": "Csatornák automatikus címkézése", + "playlist_sync": "Lejátszási listák szinkronizálása", + "working": "Dolgozik…" }, "maintenance": { "title": "Karbantartás", diff --git a/frontend/src/lib/useLiveQuery.ts b/frontend/src/lib/useLiveQuery.ts index 121045b..9571337 100644 --- a/frontend/src/lib/useLiveQuery.ts +++ b/frontend/src/lib/useLiveQuery.ts @@ -8,14 +8,22 @@ import { useQuery, type QueryKey } from "@tanstack/react-query"; export function useLiveQuery( key: QueryKey, queryFn: () => Promise, - opts: { intervalMs?: number; enabled?: boolean } = {} + // intervalMs may be a function of the latest data, so the poll cadence can adapt to it + // (e.g. poll faster while a job is running). react-query re-evaluates it after each fetch + // against fresh data — the right place for this, vs. a React state toggle that can lag or + // stall the live updates. + opts: { intervalMs?: number | ((data: T | undefined) => number); enabled?: boolean } = {} ) { const { intervalMs = 4000, enabled = true } = opts; return useQuery({ queryKey: key, queryFn, enabled, - refetchInterval: enabled ? intervalMs : false, + refetchInterval: !enabled + ? false + : typeof intervalMs === "function" + ? (query) => intervalMs(query.state.data as T | undefined) + : intervalMs, refetchIntervalInBackground: false, staleTime: 0, });