Merge bug/scheduler-progress-visibility: live progress for any scheduler job + parallel RSS polling

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}
 

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}

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
+    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:
-            new += poll_rss_channel(db, channel)
+                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 poll failed for channel %s", channel.id)
+                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)}
 
 

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", {})

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<SchedulerJob["progress"]> }) {
+// 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 (
     <div className="mt-1.5">
       <div className="flex items-center justify-between text-[11px] text-muted mb-0.5">
         <span className="truncate">{phase}</span>
         <span className="tabular-nums shrink-0">
-          {p.total != null
+          {p?.total != null
             ? `${p.current.toLocaleString()} / ${p.total.toLocaleString()}`
-            : p.current.toLocaleString()}
+            : p
+              ? p.current.toLocaleString()
+              : ""}
         </span>
       </div>
       <div className="h-1.5 rounded-full bg-border overflow-hidden">
@@ -184,7 +192,7 @@ function JobRow({
             <span> · {job.last_result}</span>
           ) : null}
         </div>
-        {job.running && job.progress && <JobProgress p={job.progress} />}
+        {job.running && <JobProgress p={job.progress} />}
       </div>
       <div className="shrink-0 text-right text-[11px] text-muted tabular-nums">
         {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<SchedulerStatus>(["scheduler"], api.schedulerStatus, {
-    intervalMs: POLL_MS,
+    intervalMs: (d) => (d?.jobs?.some((j) => j.running) ? FAST_POLL_MS : POLL_MS),
   });
   const data = q.data;
 

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",

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",

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",

frontend/src/lib/useLiveQuery.ts

@@ -8,14 +8,22 @@ import { useQuery, type QueryKey } from "@tanstack/react-query";
 export function useLiveQuery<T>(
   key: QueryKey,
   queryFn: () => Promise<T>,
-  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,
   });