From 8b01ada68a55e1d95bb617f13bf84f501a00c32a Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sun, 14 Jun 2026 13:55:59 -0700 Subject: [PATCH] video: pause enrichment workers during library scans, resume after MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every scan (incremental / full / deep, both entry points) now steps the enrichment workers aside to cut DB lock contention — same as music. Mirrors music's contract exactly: pause ONLY workers that were running (a user's manual pause is left alone), track which we paused, and resume just those in a finally so success, cancel, and error all clean up. Auto-pause is transient (persist=False) so it never leaks into the saved _paused flag. Hooks are injected by get_video_scanner; the scanner is inert without them (tests build it directly, no engine spun up). Isolated to the video side. --- core/video/enrichment/engine.py | 28 ++++++++++++++++++ core/video/enrichment/worker.py | 10 ++++--- core/video/scanner.py | 46 +++++++++++++++++++++++++++-- tests/test_video_enrichment.py | 25 ++++++++++++++++ tests/test_video_scanner.py | 52 +++++++++++++++++++++++++++++++++ 5 files changed, 155 insertions(+), 6 deletions(-) diff --git a/core/video/enrichment/engine.py b/core/video/enrichment/engine.py index f058f413..0e05f40f 100644 --- a/core/video/enrichment/engine.py +++ b/core/video/enrichment/engine.py @@ -37,6 +37,34 @@ class VideoEnrichmentEngine: for w in self.workers.values(): w.stop() + # ── scan coupling ───────────────────────────────────────────────────────── + # While a library scan runs, the enrichment workers step aside to cut DB lock + # contention — exactly like the music side. We pause ONLY workers that were + # actually running (never a user's manual pause) and remember which, so the + # post-scan resume can't un-pause something the user deliberately paused. The + # auto-pause is transient (persist=False) so it never leaks into the saved + # _paused flag and survives a restart as a "real" pause. + def pause_for_scan(self) -> set: + self._scan_paused = set() + for service, w in self.workers.items(): + if not w.paused: + w.pause(persist=False) + self._scan_paused.add(service) + if self._scan_paused: + logger.info("video enrichment: paused %s for library scan", + ", ".join(sorted(self._scan_paused))) + return self._scan_paused + + def resume_after_scan(self) -> None: + for service in getattr(self, "_scan_paused", set()): + w = self.workers.get(service) + if w: + w.resume(persist=False) + if getattr(self, "_scan_paused", None): + logger.info("video enrichment: resumed %s after library scan", + ", ".join(sorted(self._scan_paused))) + self._scan_paused = set() + def worker(self, service): return self.workers.get(service) diff --git a/core/video/enrichment/worker.py b/core/video/enrichment/worker.py index d88eae92..efae9328 100644 --- a/core/video/enrichment/worker.py +++ b/core/video/enrichment/worker.py @@ -50,13 +50,15 @@ class VideoEnrichmentWorker: self._thread.join(timeout=1.0) self.running = False - def pause(self): + def pause(self, persist=True): self.paused = True - self._persist_paused() + if persist: + self._persist_paused() - def resume(self): + def resume(self, persist=True): self.paused = False - self._persist_paused() + if persist: + self._persist_paused() def _persist_paused(self): # Survives restart, like music's _enrichment_paused config flag. diff --git a/core/video/scanner.py b/core/video/scanner.py index 9c2fac7d..3a8cf9e7 100644 --- a/core/video/scanner.py +++ b/core/video/scanner.py @@ -41,12 +41,16 @@ INCREMENTAL_MIN_LIBRARY = 50 class VideoLibraryScanner: """Reads the active media server and upserts movies/shows into video.db.""" - def __init__(self, db): + def __init__(self, db, pause_workers=None, resume_workers=None): self.db = db self._lock = threading.Lock() self._status = {"state": "idle"} self._thread = None self._cancel = False + # Optional hooks: pause enrichment workers while a scan runs, resume after + # (injected by get_video_scanner; left None in tests so no engine spins up). + self._pause_workers = pause_workers + self._resume_workers = resume_workers def get_status(self) -> dict: with self._lock: @@ -101,7 +105,30 @@ class VideoLibraryScanner: movies=movies, shows=shows, episodes=episodes) logger.info("Video scan cancelled at %d movies, %d shows", movies, shows) + def _pause_for_scan(self) -> bool: + """Pause enrichment workers for the duration of the scan. Best-effort — + a failure here must never abort the scan.""" + if not self._pause_workers: + return False + try: + self._pause_workers() + return True + except Exception: + logger.debug("video scan: pausing enrichment workers failed", exc_info=True) + return False + + def _resume_after_scan(self) -> None: + if not self._resume_workers: + return + try: + self._resume_workers() + except Exception: + logger.debug("video scan: resuming enrichment workers failed", exc_info=True) + def _run(self, source_factory, mode: str = "full") -> None: + # Enrichment steps aside for the scan (all modes, both entry points), and + # the finally guarantees it resumes on success, cancel, or error. + paused = self._pause_for_scan() try: source = source_factory() if source is None: @@ -203,6 +230,9 @@ class VideoLibraryScanner: except Exception as e: # noqa: BLE001 - report any failure to the UI logger.exception("Video library scan failed") self._set(state="error", phase="failed", error=str(e)) + finally: + if paused: + self._resume_after_scan() # Module-level singleton, bound to the (single) video DB. @@ -210,10 +240,22 @@ _scanner = None _scanner_lock = threading.Lock() +def _engine_pause_for_scan() -> None: + from core.video.enrichment.engine import get_video_enrichment_engine + get_video_enrichment_engine().pause_for_scan() + + +def _engine_resume_after_scan() -> None: + from core.video.enrichment.engine import get_video_enrichment_engine + get_video_enrichment_engine().resume_after_scan() + + def get_video_scanner(db) -> VideoLibraryScanner: global _scanner if _scanner is None: with _scanner_lock: if _scanner is None: - _scanner = VideoLibraryScanner(db) + _scanner = VideoLibraryScanner( + db, pause_workers=_engine_pause_for_scan, + resume_workers=_engine_resume_after_scan) return _scanner diff --git a/tests/test_video_enrichment.py b/tests/test_video_enrichment.py index 41dc8f74..27998377 100644 --- a/tests/test_video_enrichment.py +++ b/tests/test_video_enrichment.py @@ -120,6 +120,31 @@ def test_engine_restores_paused_workers_on_build(db): assert eng.worker("tmdb").paused is False +def test_scan_pause_resumes_only_what_it_paused(db): + eng = VideoEnrichmentEngine(db, {"tmdb": FakeClient(None), "tvdb": FakeClient(None)}) + # User manually paused tvdb (persisted); tmdb is running. + eng.worker("tvdb").pause() + assert eng.worker("tvdb").paused and not eng.worker("tmdb").paused + + paused = eng.pause_for_scan() + assert paused == {"tmdb"} # only the running one + assert eng.worker("tmdb").paused and eng.worker("tvdb").paused + + eng.resume_after_scan() + assert not eng.worker("tmdb").paused # we paused it → resumed + assert eng.worker("tvdb").paused # user's pause left alone + + +def test_scan_pause_does_not_persist(db): + eng = VideoEnrichmentEngine(db, {"tmdb": FakeClient(None)}) + eng.pause_for_scan() + # Transient: the durable flag is untouched, so a restart mid-scan won't + # restore the worker as "paused". + assert (db.get_setting("tmdb_paused") or "") != "1" + eng.resume_after_scan() + assert (db.get_setting("tmdb_paused") or "") != "1" + + def test_enrichment_package_imports_nothing_from_music(): base = Path(__file__).resolve().parent.parent / "core" / "video" / "enrichment" for py in base.glob("*.py"): diff --git a/tests/test_video_scanner.py b/tests/test_video_scanner.py index 40450eda..456b170d 100644 --- a/tests/test_video_scanner.py +++ b/tests/test_video_scanner.py @@ -49,6 +49,58 @@ def test_scan_sync_populates_library(db): assert (lib["movies"], lib["shows"], lib["episodes"]) == (1, 1, 1) +def test_scan_pauses_and_resumes_enrichment_workers(db): + events = [] + scanner = VideoLibraryScanner( + db, pause_workers=lambda: events.append("pause"), + resume_workers=lambda: events.append("resume")) + scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], [])) + assert events == ["pause", "resume"] # paused first, resumed after + + +def test_scan_resumes_enrichment_even_on_error(db): + events = [] + scanner = VideoLibraryScanner( + db, pause_workers=lambda: events.append("pause"), + resume_workers=lambda: events.append("resume")) + + def boom(): + raise RuntimeError("server blew up") + + scanner.scan_sync(boom) # source_factory raises + assert scanner.get_status()["state"] == "error" + assert events == ["pause", "resume"] # resumed despite the failure + + +def test_scan_resumes_enrichment_on_cancel(db): + events = [] + scanner = VideoLibraryScanner( + db, pause_workers=lambda: events.append("pause"), + resume_workers=lambda: events.append("resume")) + + class S: + server_name = "plex" + def counts(self, incremental=False): + return {"movies": 5, "shows": 0} + def iter_movies(self, incremental=False): + for i in range(5): + scanner._cancel = True # cancel mid-iteration + yield {"server_id": "m%d" % i, "title": "M%d" % i} + def iter_shows(self, incremental=False): + return iter([]) + + scanner.scan_sync(lambda: S()) + assert scanner.get_status()["state"] == "cancelled" + assert events == ["pause", "resume"] # resumed after cancel too + + +def test_scanner_with_no_hooks_does_not_pause(db): + # Default construction (as in tests / when no engine wired) must be inert. + scanner = VideoLibraryScanner(db) + st = scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], [])) + assert st["state"] == "done" # scans fine without hooks + + def test_deep_scan_prunes_removed_items(db): scanner = VideoLibraryScanner(db) scanner.scan_sync(lambda: FakeSource(