From 79648a4f5f6d02af466da76b063fc7bfb0974ef5 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Mon, 29 Jun 2026 09:24:41 -0700 Subject: [PATCH] import: async staging scan so a large library doesn't time out the page (#947, phase 1 backend) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A whole-library migration (ramonskie copied his Lidarr library into /staging) makes the synchronous staging scan walk + tag-read tens of thousands of files INSIDE the GET request, blowing past gunicorn's 120s timeout — and because the killed request never warms the cache, every reload re-times-out. Moves the SAME scan off the request thread; the page reports progress instead of hanging. - _scan_staging_records gains an optional `progress` param (additive; default None = unchanged). Refactored to two passes: a fast walk to collect the audio-file list (total), then the slow tag-read loop updating scanned. A generation guard stops a scan that finishes AFTER an import from committing stale records. - ensure_background_staging_scan(path): idempotent background runner filling the existing cache. - get_staging_records_or_status(): warm cache or a scan that finishes within a ~3s grace → records (so small/normal folders still answer in one request, no UX change); else ("scanning", progress). A scan error is re-raised so the endpoints log + return it exactly as before. - /staging/files|groups|hints return {scanning, progress} when the scan is still running instead of blocking; new lightweight /staging/scan-status for cheap progress polling. Single source preserved (same scan + cache, just off the request thread). 13 new tests (progress, idempotent ensure, grace ready-vs-scanning, generation guard discards stale, endpoint scanning shape, error contract, status ready/cold); full import suite 626 green; ruff clean. Next: phase 2 — the React import page polls scan-status + shows a progress bar, then renders. --- core/imports/routes.py | 185 ++++++++++++++++++++--- tests/imports/test_staging_scan_async.py | 163 ++++++++++++++++++++ web_server.py | 7 + 3 files changed, 333 insertions(+), 22 deletions(-) create mode 100644 tests/imports/test_staging_scan_async.py diff --git a/core/imports/routes.py b/core/imports/routes.py index 09043aed..77448c62 100644 --- a/core/imports/routes.py +++ b/core/imports/routes.py @@ -8,7 +8,7 @@ import time import uuid from concurrent.futures import as_completed from dataclasses import dataclass -from typing import Any, Callable, Dict +from typing import Any, Callable, Dict, Optional from core.imports.album import build_album_import_context, build_album_import_match_payload, resolve_album_artist_context from core.imports.context import get_import_context_artist, get_import_track_info, normalize_import_context @@ -84,12 +84,124 @@ class ImportRouteRuntime: _STAGING_SCAN_LOCK = threading.Lock() _STAGING_SCAN_TTL = 6.0 # seconds — covers the page-open burst; re-scans after _staging_scan_cache: Dict[str, Any] = {"path": None, "ts": 0.0, "records": None} +# Bumped by invalidate_staging_scan_cache() so a background scan that finishes after an +# import doesn't re-commit stale (pre-import) records (see the generation guard above). +_staging_scan_generation: Dict[str, int] = {"value": 0} + +# Background-scan plumbing: a large staging folder (whole-library migration, #947) makes +# the synchronous scan exceed gunicorn's 120s request timeout. The runner moves the SAME +# scan off the request thread; the endpoints report progress instead of blocking. +_staging_scan_status: Dict[str, Any] = { + "status": "idle", "scanned": 0, "total": 0, "path": None, "error": None, +} +_staging_scan_status_lock = threading.Lock() -def _scan_staging_records(runtime: ImportRouteRuntime, staging_path: str) -> list[Dict[str, Any]]: +def _staging_cache_hit(staging_path: str) -> Optional[list]: + """The cached records for ``staging_path`` if still fresh, else None (no scan triggered).""" + c = _staging_scan_cache + if (c["records"] is not None and c["path"] == staging_path + and (time.time() - c["ts"]) < _STAGING_SCAN_TTL): + return c["records"] + return None + + +def ensure_background_staging_scan(runtime: ImportRouteRuntime, staging_path: str) -> None: + """Start a background scan for ``staging_path`` unless the cache is warm or a scan for + this path is already running. Idempotent — safe to call on every request.""" + if _staging_cache_hit(staging_path) is not None: + return + with _staging_scan_status_lock: + if (_staging_scan_status["status"] == "scanning" + and _staging_scan_status["path"] == staging_path): + return + _staging_scan_status.update({"status": "scanning", "scanned": 0, "total": 0, + "path": staging_path, "error": None}) + + def _run() -> None: + try: + _scan_staging_records(runtime, staging_path, progress=_staging_scan_status) + with _staging_scan_status_lock: + if _staging_scan_status["path"] == staging_path: + _staging_scan_status["status"] = "done" + except Exception as exc: # noqa: BLE001 — surface any scan error to the poller + with _staging_scan_status_lock: + _staging_scan_status.update({"status": "error", "error": str(exc)}) + + threading.Thread(target=_run, name="staging-scan", daemon=True).start() + + +def get_staging_records_or_status(runtime: ImportRouteRuntime, staging_path: str, + *, grace_seconds: float = 3.0) -> tuple[str, Any]: + """Non-blocking staging access for the page endpoints. Returns ``("ready", records)`` + when the cache is warm or the scan completes within ``grace_seconds`` (so small/normal + folders still answer in a single request), otherwise ``("scanning", status_dict)`` after + making sure a background scan is running.""" + records = _staging_cache_hit(staging_path) + if records is not None: + return ("ready", records) + ensure_background_staging_scan(runtime, staging_path) + deadline = time.time() + max(0.0, grace_seconds) + while True: + records = _staging_cache_hit(staging_path) + if records is not None: + return ("ready", records) + with _staging_scan_status_lock: + status = dict(_staging_scan_status) + if status.get("status") == "error": + return ("error", status) + if time.time() >= deadline: + return ("scanning", status) + time.sleep(0.05) + + +def _records_or_scanning_payload(runtime: ImportRouteRuntime, staging_path: str): + """Shared helper for the page endpoints: returns ``(records, None)`` when the scan is + ready, or ``(None, payload)`` when a background scan is still running — the caller + returns that payload so the page polls + shows progress instead of blocking/timing out. + + A scan error is re-raised so the endpoint's own try/except logs + returns it exactly as + when the scan ran inline (preserves the existing error contract).""" + state, val = get_staging_records_or_status(runtime, staging_path) + if state == "error": + raise RuntimeError(val.get("error") or "staging scan failed") + if state == "scanning": + return None, {"success": True, "scanning": True, + "progress": {"scanned": val.get("scanned", 0), + "total": val.get("total", 0)}} + return val, None + + +def staging_scan_status(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: + """Lightweight, instant scan-progress poll for the page (no grace-wait, no file I/O) — + ``ready`` true once the cache is warm and the files/groups/hints calls will answer fast.""" + try: + staging_path = runtime.get_staging_path() + except Exception as exc: + return {"success": False, "error": str(exc)}, 500 + with _staging_scan_status_lock: + st = dict(_staging_scan_status) + return { + "success": True, + "ready": _staging_cache_hit(staging_path) is not None, + "status": st.get("status", "idle"), + "scanned": st.get("scanned", 0), + "total": st.get("total", 0), + "error": st.get("error"), + }, 200 + + +def _scan_staging_records(runtime: ImportRouteRuntime, staging_path: str, + *, progress: Optional[Dict[str, Any]] = None) -> list[Dict[str, Any]]: """Walk staging + read each audio file's tags ONCE, returning per-file records that staging files/groups/hints all derive from. Briefly cached + locked so the - page-open trio shares a single scan rather than each re-walking and re-reading.""" + page-open trio shares a single scan rather than each re-walking and re-reading. + + ``progress`` (optional, default None = unchanged behaviour) is a dict the scan + updates live with ``total`` (audio-file count, from a fast first pass) and ``scanned`` + (tag-reads done so far) so a background runner can report progress. A generation guard + keeps a scan that finishes AFTER an import (which bumped ``_staging_scan_generation``) + from committing stale records to the cache.""" now = time.time() cached = _staging_scan_cache if (cached["records"] is not None and cached["path"] == staging_path @@ -103,33 +215,50 @@ def _scan_staging_records(runtime: ImportRouteRuntime, staging_path: str) -> lis and (now - cached["ts"]) < _STAGING_SCAN_TTL): return cached["records"] - records: list[Dict[str, Any]] = [] + start_generation = _staging_scan_generation["value"] + + # Pass 1 (fast): collect the audio-file list — no tag I/O — so we know the total. + audio_files: list[tuple[str, str, Optional[str]]] = [] if os.path.isdir(staging_path): for root, _dirs, filenames in os.walk(staging_path): rel_dir = os.path.relpath(root, staging_path) top_folder = rel_dir.split(os.sep)[0] if rel_dir != "." else None for fname in filenames: - ext = os.path.splitext(fname)[1].lower() - if ext not in AUDIO_EXTENSIONS: - continue - full_path = os.path.join(root, fname) - rel_path = os.path.relpath(full_path, staging_path) - meta = runtime.read_staging_file_metadata(full_path, rel_path) - records.append({ - "filename": fname, "rel_path": rel_path, "full_path": full_path, - "extension": ext, "title": meta["title"], "album": meta["album"], - "artist": meta["artist"], "albumartist": meta["albumartist"], - "track_number": meta["track_number"], "disc_number": meta["disc_number"], - "top_folder": top_folder, - }) + if os.path.splitext(fname)[1].lower() in AUDIO_EXTENSIONS: + audio_files.append((root, fname, top_folder)) + if progress is not None: + progress["total"] = len(audio_files) + progress["scanned"] = 0 - _staging_scan_cache.update({"path": staging_path, "ts": time.time(), "records": records}) + # Pass 2 (slow): read each file's tags, updating progress as we go. + records: list[Dict[str, Any]] = [] + for root, fname, top_folder in audio_files: + full_path = os.path.join(root, fname) + rel_path = os.path.relpath(full_path, staging_path) + meta = runtime.read_staging_file_metadata(full_path, rel_path) + records.append({ + "filename": fname, "rel_path": rel_path, "full_path": full_path, + "extension": os.path.splitext(fname)[1].lower(), + "title": meta["title"], "album": meta["album"], + "artist": meta["artist"], "albumartist": meta["albumartist"], + "track_number": meta["track_number"], "disc_number": meta["disc_number"], + "top_folder": top_folder, + }) + if progress is not None: + progress["scanned"] += 1 + + # Generation guard: if an import invalidated the cache mid-scan, these records are + # stale — return them to this caller but do NOT commit them as the shared cache. + if _staging_scan_generation["value"] == start_generation: + _staging_scan_cache.update({"path": staging_path, "ts": time.time(), "records": records}) return records def invalidate_staging_scan_cache() -> None: """Drop the cached staging scan (call after an import moves/removes files so the - next files/groups/hints request reflects the new state immediately).""" + next files/groups/hints request reflects the new state immediately). Also bumps the + scan generation so an in-flight background scan won't re-commit pre-import records.""" + _staging_scan_generation["value"] += 1 _staging_scan_cache.update({"path": None, "ts": 0.0, "records": None}) @@ -139,6 +268,10 @@ def staging_files(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: staging_path = runtime.get_staging_path() os.makedirs(staging_path, exist_ok=True) + records, scanning = _records_or_scanning_payload(runtime, staging_path) + if scanning is not None: + return scanning, 200 + files = [ { "filename": r["filename"], @@ -151,7 +284,7 @@ def staging_files(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: "disc_number": r["disc_number"], "extension": r["extension"], } - for r in _scan_staging_records(runtime, staging_path) + for r in records ] files.sort(key=lambda f: f["filename"].lower()) @@ -168,8 +301,12 @@ def staging_groups(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: if not os.path.isdir(staging_path): return {"success": True, "groups": []}, 200 + records, scanning = _records_or_scanning_payload(runtime, staging_path) + if scanning is not None: + return scanning, 200 + album_groups = {} - for r in _scan_staging_records(runtime, staging_path): + for r in records: album = r["album"] artist = r["albumartist"] or r["artist"] if not album or not artist: @@ -215,9 +352,13 @@ def staging_hints(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: if not os.path.isdir(staging_path): return {"success": True, "hints": []}, 200 + records, scanning = _records_or_scanning_payload(runtime, staging_path) + if scanning is not None: + return scanning, 200 + tag_albums = {} folder_hints = {} - for r in _scan_staging_records(runtime, staging_path): + for r in records: if r["top_folder"]: folder_hints[r["top_folder"]] = folder_hints.get(r["top_folder"], 0) + 1 diff --git a/tests/imports/test_staging_scan_async.py b/tests/imports/test_staging_scan_async.py new file mode 100644 index 00000000..9a89900e --- /dev/null +++ b/tests/imports/test_staging_scan_async.py @@ -0,0 +1,163 @@ +"""Async/background staging scan (#947): a whole-library migration makes the synchronous +scan exceed gunicorn's 120s timeout. The runner moves the SAME scan off the request thread +with progress + a generation guard. Metadata reads are injected so no real audio is needed.""" + +import os +import time +import types + +import pytest + +import core.imports.routes as routes + + +def _meta(_full, _rel): + return {"title": "t", "album": "Alb", "artist": "Art", "albumartist": "Art", + "track_number": None, "disc_number": None} + + +def _runtime(read=_meta): + return types.SimpleNamespace(read_staging_file_metadata=read) + + +def _staging(tmp_path, n=3, subdir="Artist/Album"): + d = tmp_path / "staging" + for part in subdir.split("/"): + d = d / part + d.mkdir(parents=True, exist_ok=True) + for i in range(n): + (d / f"{i:02d}.flac").write_text("x") + return str(tmp_path / "staging") + + +@pytest.fixture(autouse=True) +def _reset(): + routes.invalidate_staging_scan_cache() + routes._staging_scan_status.update({"status": "idle", "scanned": 0, "total": 0, + "path": None, "error": None}) + yield + routes.invalidate_staging_scan_cache() + + +def _await_done(timeout=5.0): + end = time.time() + timeout + while time.time() < end and routes._staging_scan_status["status"] == "scanning": + time.sleep(0.03) + + +def test_scan_reports_progress(tmp_path): + sp = _staging(tmp_path, 3) + prog = {} + recs = routes._scan_staging_records(_runtime(), sp, progress=prog) + assert len(recs) == 3 + assert prog["total"] == 3 and prog["scanned"] == 3 + + +def test_default_scan_behaviour_unchanged(tmp_path): + sp = _staging(tmp_path, 2) + assert len(routes._scan_staging_records(_runtime(), sp)) == 2 # no progress arg = as before + + +def test_accessor_ready_for_small_folder(tmp_path): + sp = _staging(tmp_path, 2) + state, val = routes.get_staging_records_or_status(_runtime(), sp, grace_seconds=3.0) + assert state == "ready" and len(val) == 2 + + +def test_accessor_scanning_when_scan_exceeds_grace(tmp_path): + sp = _staging(tmp_path, 2) + + def slow(_f, _r): + time.sleep(0.4) + return _meta(_f, _r) + + state, val = routes.get_staging_records_or_status(_runtime(slow), sp, grace_seconds=0.1) + assert state == "scanning" + assert val["status"] == "scanning" and val["total"] in (0, 2) + _await_done() + + +def test_ensure_scan_is_idempotent(tmp_path): + sp = _staging(tmp_path, 2) + calls = {"n": 0} + + def counting(_f, _r): + calls["n"] += 1 + time.sleep(0.15) + return _meta(_f, _r) + + rt = _runtime(counting) + routes.ensure_background_staging_scan(rt, sp) + routes.ensure_background_staging_scan(rt, sp) # must NOT start a second scan + _await_done() + assert calls["n"] == 2 # 2 files read once, not 4 + + +def test_generation_guard_discards_stale_records(tmp_path): + sp = _staging(tmp_path, 2) + + def read_then_import(_f, _r): + routes.invalidate_staging_scan_cache() # simulate an import landing mid-scan + return _meta(_f, _r) + + recs = routes._scan_staging_records(_runtime(read_then_import), sp) + assert len(recs) == 2 # caller still gets its records + assert routes._staging_scan_cache["records"] is None # but stale set NOT committed to cache + + +def _full_runtime(staging_path, read=_meta): + return types.SimpleNamespace( + get_staging_path=lambda: staging_path, + read_staging_file_metadata=read, + logger=types.SimpleNamespace(error=lambda *a, **k: None), + ) + + +def test_helper_passes_records_through_when_ready(monkeypatch): + monkeypatch.setattr(routes, 'get_staging_records_or_status', + lambda rt, sp: ("ready", [{"x": 1}])) + records, scanning = routes._records_or_scanning_payload(_runtime(), "/x") + assert scanning is None and records == [{"x": 1}] + + +def test_helper_builds_scanning_payload(monkeypatch): + monkeypatch.setattr(routes, 'get_staging_records_or_status', + lambda rt, sp: ("scanning", {"scanned": 5, "total": 20, "status": "scanning"})) + records, scanning = routes._records_or_scanning_payload(_runtime(), "/x") + assert records is None + assert scanning == {"success": True, "scanning": True, + "progress": {"scanned": 5, "total": 20}} + + +def test_staging_files_endpoint_ready(tmp_path): + payload, status = routes.staging_files(_full_runtime(_staging(tmp_path, 2))) + assert status == 200 and payload["success"] and len(payload["files"]) == 2 + + +def test_staging_files_endpoint_returns_scanning(tmp_path, monkeypatch): + monkeypatch.setattr(routes, 'get_staging_records_or_status', + lambda r, p: ("scanning", {"scanned": 3, "total": 10})) + payload, status = routes.staging_files(_full_runtime(_staging(tmp_path, 2))) + assert status == 200 and payload.get("scanning") is True + assert payload["progress"] == {"scanned": 3, "total": 10} + + +def test_staging_groups_endpoint_returns_scanning(tmp_path, monkeypatch): + monkeypatch.setattr(routes, 'get_staging_records_or_status', + lambda r, p: ("scanning", {"scanned": 1, "total": 9})) + payload, status = routes.staging_groups(_full_runtime(_staging(tmp_path, 2))) + assert payload.get("scanning") is True + + +def test_scan_status_ready_after_warm(tmp_path): + sp = _staging(tmp_path, 2) + routes._scan_staging_records(_runtime(), sp) # warm the cache + payload, status = routes.staging_scan_status(_full_runtime(sp)) + assert status == 200 and payload["success"] and payload["ready"] is True + + +def test_scan_status_not_ready_when_cold(tmp_path): + sp = _staging(tmp_path, 2) # cold (autouse reset) + payload, _ = routes.staging_scan_status(_full_runtime(sp)) + assert payload["ready"] is False + assert "scanned" in payload and "total" in payload diff --git a/web_server.py b/web_server.py index df083341..a45811cd 100644 --- a/web_server.py +++ b/web_server.py @@ -177,6 +177,7 @@ from core.imports.routes import singles_process as _import_singles_process from core.imports.routes import staging_files as _import_staging_files from core.imports.routes import staging_groups as _import_staging_groups from core.imports.routes import staging_hints as _import_staging_hints +from core.imports.routes import staging_scan_status as _import_staging_scan_status from core.imports.routes import staging_suggestions as _import_staging_suggestions from core.imports.paths import build_final_path_for_track as _build_final_path_for_track from core.imports.pipeline import build_import_pipeline_runtime as _build_import_pipeline_runtime @@ -37700,6 +37701,12 @@ def import_staging_groups(): return jsonify(payload), status +@app.route('/api/import/staging/scan-status', methods=['GET']) +def import_staging_scan_status(): + payload, status = _import_staging_scan_status(_build_import_route_runtime()) + return jsonify(payload), status + + @app.route('/api/import/staging/hints', methods=['GET']) def import_staging_hints(): payload, status = _import_staging_hints(_build_import_route_runtime())