import: async staging scan so a large library doesn't time out the page (#947, phase 1 backend)
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.
This commit is contained in:
BoulderBadgeDad
parent
b05b641521
commit
79648a4f5f
3 changed files with 333 additions and 22 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
163
tests/imports/test_staging_scan_async.py
Normal file
163
tests/imports/test_staging_scan_async.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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())
|
||||
|
|
|
|||
Loading…
Reference in a new issue