youtube fulfillment: orphan reaper for restart-stalled downloads

closes the real gap: a restart kills the yt-dlp worker threads but leaves their
rows at 'downloading', which the pump counts as busy -> the queue wedges and they
never finish.

track live worker dl_ids in _active_worker_ids (added at worker start, dropped in
finally). requeue_orphaned_youtube() puts any 'downloading' youtube row with no
live worker back to 'queued' so the pump re-runs it. after a restart the set is
empty, so all stuck rows recover; during normal operation active ones are
protected (no false positives, no timestamp guessing). the hourly drain calls it
before pumping + logs the count. seam-tested.

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 []

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

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
 

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 = {}