diff --git a/core/automation/handlers/video_process_youtube_wishlist.py b/core/automation/handlers/video_process_youtube_wishlist.py index df8d2dbf..6bc16189 100644 --- a/core/automation/handlers/video_process_youtube_wishlist.py +++ b/core/automation/handlers/video_process_youtube_wishlist.py @@ -91,6 +91,13 @@ def _default_start_next() -> Any: return start_next_queued(get_video_db) +def _default_reap() -> int: + """Recover downloads orphaned by a restart (stuck 'downloading', no live worker).""" + from api.video import get_video_db + from core.video.youtube_download import requeue_orphaned_youtube + return requeue_orphaned_youtube(get_video_db) + + def auto_video_process_youtube_wishlist( config: Dict[str, Any], deps: AutomationDeps, @@ -101,6 +108,7 @@ def auto_video_process_youtube_wishlist( running_count: Optional[Callable[[], int]] = None, enqueue: Optional[Callable[[Dict[str, Any], str], Any]] = None, start_next: Optional[Callable[[], Any]] = None, + reap: Optional[Callable[[], int]] = None, ) -> Dict[str, Any]: """Queue the whole YouTube wishlist for download and start up to ``max_concurrent`` now. @@ -111,6 +119,7 @@ def auto_video_process_youtube_wishlist( running_count = running_count or _default_running_count enqueue = enqueue or _default_enqueue start_next = start_next or _default_start_next + reap = reap or _default_reap automation_id = config.get('_automation_id') max_concurrent = max(1, int(config.get('max_concurrent', 3) or 3)) @@ -125,6 +134,13 @@ def auto_video_process_youtube_wishlist( return {'status': 'completed', 'queued': 0, 'started': 0, 'running': 0, 'skipped': 'no_youtube_folder', '_manages_own_progress': True} + # Recover any downloads orphaned by a restart (stuck 'downloading', no worker) so + # they don't wedge the concurrency count — they go back to 'queued' and re-run. + recovered = int(reap() or 0) + if recovered: + deps.update_progress(automation_id, log_type='info', + log_line='Recovered %d stalled download(s) from a restart' % recovered) + deps.update_progress(automation_id, phase='Checking the YouTube wishlist…', progress=15, log_line='Queueing new videos for download', log_type='info') wanted = fetch_wanted() or [] diff --git a/core/video/youtube_download.py b/core/video/youtube_download.py index 446f9b92..796eaf9e 100644 --- a/core/video/youtube_download.py +++ b/core/video/youtube_download.py @@ -196,11 +196,30 @@ def _pace(delay: float) -> None: time.sleep(wait) +# Download ids with a live worker thread right now. After a restart this is empty, so any +# row still marked 'downloading' is an orphan (its thread died) → the reaper re-queues it. +_active_worker_ids: set = set() + + def _spawn_worker(dl_id: Any, db_provider: Callable) -> None: threading.Thread(target=run_youtube_download, args=(dl_id, db_provider), daemon=True, name="yt-dl-%s" % dl_id).start() +def requeue_orphaned_youtube(db_provider: Callable) -> int: + """Recover YouTube downloads stuck in 'downloading' with no live worker (e.g. after a + restart killed the threads) by putting them back to 'queued' so the pump re-runs them. + A download whose worker is alive is in ``_active_worker_ids`` and is left untouched. + Returns the count recovered.""" + n = 0 + for d in (db_provider().get_active_video_downloads() or []): + if (d.get("source") == "youtube" and d.get("status") == "downloading" + and d.get("id") not in _active_worker_ids): + db_provider().update_video_download(d["id"], status="queued", progress=0) + n += 1 + return n + + def start_next_queued(db_provider: Callable) -> Any: """Claim the next queued YouTube download and start it. Returns its id, or None if the queue is empty. The wishlist pump uses this to fill slots; each finished worker calls it @@ -216,9 +235,11 @@ def start_next_queued(db_provider: Callable) -> Any: def run_youtube_download(dl_id: Any, db_provider: Callable) -> None: """Production entry: fetch the row, bind real seams, fulfil it. Called on a worker thread by the pump; on finish it starts the next queued download (one-out-one-in).""" + _active_worker_ids.add(dl_id) # mark this download as having a live worker db = db_provider() dl = db.get_video_download(dl_id) if not dl: + _active_worker_ids.discard(dl_id) start_next_queued(db_provider) # keep the queue moving even on a stale id return profile = youtube_quality.load(db) @@ -260,11 +281,12 @@ def run_youtube_download(dl_id: Any, db_provider: Callable) -> None: clear_wishlist=lambda vid: db.remove_youtube_from_wishlist("video", vid), progress_hook=_progress, cookie_opts=cookie_opts, now=_now) finally: + _active_worker_ids.discard(dl_id) # worker done — no longer protects this row start_next_queued(db_provider) # one out, one in — drain the queue __all__ = [ "youtube_fields_from_download", "plan_destination", "ydl_download_opts", "download_one", "process_youtube_download", "run_youtube_download", - "start_next_queued", + "start_next_queued", "requeue_orphaned_youtube", ] diff --git a/tests/test_video_process_youtube_wishlist.py b/tests/test_video_process_youtube_wishlist.py index 91d62359..e361e658 100644 --- a/tests/test_video_process_youtube_wishlist.py +++ b/tests/test_video_process_youtube_wishlist.py @@ -66,7 +66,7 @@ def _run(wanted, *, active=None, running=0, root="/yt", max_concurrent=3, start_ {"_automation_id": "a", "max_concurrent": max_concurrent}, deps, youtube_root=lambda: root, fetch_wanted=lambda: wanted, active_ids=lambda: list(active or []), running_count=lambda: running, - enqueue=enqueue, start_next=start_next) + enqueue=enqueue, start_next=start_next, reap=lambda: 0) return res, enq, starts["n"], deps @@ -130,7 +130,7 @@ def test_one_bad_enqueue_does_not_stop_the_rest(): {"_automation_id": "x", "max_concurrent": 5}, _Deps(), youtube_root=lambda: "/yt", fetch_wanted=lambda: [_v("a"), _v("b")], active_ids=lambda: [], running_count=lambda: 0, enqueue=enqueue, - start_next=lambda: None) + start_next=lambda: None, reap=lambda: 0) assert res["status"] == "completed" and res["queued"] == 1 # b still queued @@ -138,10 +138,23 @@ def test_top_level_error_is_caught(): def boom(): raise RuntimeError("db down") res = auto_video_process_youtube_wishlist({"_automation_id": "x"}, _Deps(), - youtube_root=lambda: "/yt", fetch_wanted=boom) + youtube_root=lambda: "/yt", fetch_wanted=boom, + reap=lambda: 0) assert res["status"] == "error" and "db down" in res["error"] +def test_reaper_runs_and_is_reported(): + # the drain recovers restart-orphaned downloads before pumping, and logs the count + deps = _Deps() + res = auto_video_process_youtube_wishlist( + {"_automation_id": "a", "max_concurrent": 3}, deps, + youtube_root=lambda: "/yt", fetch_wanted=lambda: [], active_ids=lambda: [], + running_count=lambda: 0, enqueue=lambda v, r: 1, start_next=lambda: None, + reap=lambda: 2) + assert res["status"] == "completed" + assert any("Recovered 2 stalled" in (p.get("log_line") or "") for p in deps.progress) + + # ── the DB queue methods ────────────────────────────────────────────────────── from database.video_database import VideoDatabase # noqa: E402 diff --git a/tests/test_youtube_download.py b/tests/test_youtube_download.py index 7c47a9aa..f15e1b5a 100644 --- a/tests/test_youtube_download.py +++ b/tests/test_youtube_download.py @@ -137,6 +137,34 @@ def test_process_failure_archives_but_keeps_the_wish(): assert calls["unwish"] == [] # wish kept so it can retry later +def test_requeue_orphaned_youtube_recovers_only_dead_downloads(): + """After a restart no worker threads survive, so any 'downloading' YouTube row is an + orphan → back to 'queued'. A row whose worker is still alive (in _active_worker_ids) and + non-youtube / non-downloading rows are left alone.""" + updates = [] + + class _DB: + def get_active_video_downloads(self): + return [ + {"id": 1, "source": "youtube", "status": "downloading"}, # orphan → requeue + {"id": 2, "source": "youtube", "status": "downloading"}, # live worker → keep + {"id": 3, "source": "youtube", "status": "queued"}, # not downloading → keep + {"id": 4, "source": "soulseek", "status": "downloading"}, # not youtube → keep + ] + + def update_video_download(self, dl_id, **kw): + updates.append((dl_id, kw)) + + ytd._active_worker_ids.clear() + ytd._active_worker_ids.add(2) # id 2 has a live worker + try: + n = ytd.requeue_orphaned_youtube(lambda: _DB()) + finally: + ytd._active_worker_ids.clear() + assert n == 1 + assert updates == [(1, {"status": "queued", "progress": 0})] # only the orphan + + def test_process_passes_the_organised_dir_to_the_downloader(): seen = {}