video: pause enrichment workers during library scans, resume after
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 <service>_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.
This commit is contained in:
BoulderBadgeDad
parent
daecf9dfae
commit
8b01ada68a
5 changed files with 155 additions and 6 deletions
|
|
@ -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
|
||||
# <service>_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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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 <service>_enrichment_paused config flag.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"):
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
Loading…
Reference in a new issue