From f3392116543d6b4386f118adf3aa0c4e80d5a787 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Thu, 30 Apr 2026 22:41:04 -0700 Subject: [PATCH 1/3] Parallelize singles-import processing with a 3-worker executor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Discord-reported (fresh.dumbledore + maintainer ack): the /api/import/singles/process route iterated staging files through a plain Python for loop. Per-file work is dominated by metadata search round-trips (Spotify/iTunes/Deezer/Discogs), so a multi- track manual import on a typical home network was painfully slow. Adds a dedicated import_singles_executor (3 workers) alongside the existing executor pool, and refactors the route to submit every file at once and aggregate results via as_completed. Worker count balances throughput against any single provider's per-source rate limits — the same shape used by missing_download_executor. Extracts the per-file pipeline into _process_single_import_file which returns a typed (status, payload) outcome: - ("ok", final_title) on success - ("error", message) for missing/malformed input or pipeline failure The worker wraps its own exceptions so a single bad file can't crash the batch; the route adds a belt-and-suspenders try/except around future.result() for any worker-level surprises. Pipeline thread-safety verified: post_process_matched_download already serializes per-file via post_process_locks (one lock per context_key — and each import gets a unique UUID context_key), DB writes serialize through SQLite's WAL + busy_timeout, metadata registry uses RLocks, no bare module-level mutable state. Adds 9 regression tests: - 4 worker-contract tests (missing file, malformed match, pipeline exception wrapping, happy-path return shape) - 2 executor-config tests (worker count, thread name prefix) - 1 integration test that proves the route actually parallelizes by checking wall-clock duration is well under sequential cost - 1 mixed-outcome aggregation test - 1 worker-crash recovery test Doesn't address the related "stops on tab close" complaint — that's a separate request-lifecycle issue that needs job_id + polling, not just parallelism. --- tests/test_import_singles_parallel.py | 302 ++++++++++++++++++++++++++ web_server.py | 158 +++++++++----- 2 files changed, 405 insertions(+), 55 deletions(-) create mode 100644 tests/test_import_singles_parallel.py diff --git a/tests/test_import_singles_parallel.py b/tests/test_import_singles_parallel.py new file mode 100644 index 00000000..ff8d1ace --- /dev/null +++ b/tests/test_import_singles_parallel.py @@ -0,0 +1,302 @@ +"""Regression tests for parallel singles-import processing. + +Discord-reported (fresh.dumbledore + maintainer ack): the +``/api/import/singles/process`` endpoint processed staging files +sequentially in a Python ``for`` loop. Per-file work is dominated by +metadata search round-trips (Spotify/iTunes/Deezer), so a +multi-track manual import on a typical home network was painfully +slow. The maintainer acknowledged needing multiple workers. + +These tests pin the new behaviour: + +- The per-file worker function exists, returns a typed outcome + ``(status, payload)``, and is safe to call concurrently from the + shared ThreadPoolExecutor. +- Successful files report ``("ok", final_title)`` so the route can + count them. +- Failed metadata resolution / bad files report ``("error", msg)``. +- A worker that raises an unexpected exception is caught by the + caller (the test verifies that behaviour through the route). +""" + +from unittest.mock import patch + +import pytest + + +# --------------------------------------------------------------------------- +# Worker contract +# --------------------------------------------------------------------------- + + +def test_worker_returns_error_for_missing_file(tmp_path) -> None: + """Files whose path doesn't exist must short-circuit with a + user-readable error, not raise — otherwise the executor's caller + can't aggregate them cleanly.""" + from web_server import _process_single_import_file + + file_info = { + 'full_path': str(tmp_path / "does-not-exist.mp3"), + 'filename': 'does-not-exist.mp3', + } + outcome, payload = _process_single_import_file(file_info) + assert outcome == "error" + assert "File not found" in payload + + +def test_worker_returns_error_for_malformed_manual_match(tmp_path) -> None: + """Manual matches missing source or id must be rejected with a + clear message rather than crashing the resolver downstream.""" + from web_server import _process_single_import_file + + audio_file = tmp_path / "track.mp3" + audio_file.write_bytes(b"fake") + + file_info = { + 'full_path': str(audio_file), + 'filename': 'track.mp3', + 'manual_match': {'source': '', 'id': ''}, + } + outcome, payload = _process_single_import_file(file_info) + assert outcome == "error" + assert "Malformed manual match" in payload + + +def test_worker_wraps_pipeline_exception_as_error(tmp_path) -> None: + """If the post-processing pipeline raises, the worker must catch + it and report ``("error", msg)`` so a single bad file doesn't + take the whole batch down via the executor's caller path.""" + from web_server import _process_single_import_file + + audio_file = tmp_path / "track.mp3" + audio_file.write_bytes(b"fake") + + file_info = { + 'full_path': str(audio_file), + 'filename': 'track.mp3', + 'title': 'Some Song', + 'artist': 'Some Artist', + } + + with patch( + "core.imports.resolution.get_single_track_import_context", + side_effect=RuntimeError("metadata service down"), + ): + outcome, payload = _process_single_import_file(file_info) + assert outcome == "error" + assert "metadata service down" in payload + + +def test_worker_returns_ok_with_resolved_title(tmp_path) -> None: + """Happy path: pipeline succeeds → ``("ok", final_title)`` so the + route can use it for the activity feed message.""" + from web_server import _process_single_import_file + + audio_file = tmp_path / "track.mp3" + audio_file.write_bytes(b"fake") + + file_info = { + 'full_path': str(audio_file), + 'filename': 'track.mp3', + 'title': 'Resolved Title', + 'artist': 'Resolved Artist', + } + + fake_resolved = { + 'context': { + 'artist': {'name': 'Resolved Artist'}, + 'track_info': {'name': 'Resolved Title'}, + 'album': {}, + 'original_search_result': { + 'title': 'Resolved Title', + 'artist': 'Resolved Artist', + 'clean_title': 'Resolved Title', + 'clean_artist': 'Resolved Artist', + 'clean_album': '', + 'album': '', + }, + }, + 'source': 'spotify', + } + + with patch( + "core.imports.resolution.get_single_track_import_context", + return_value=fake_resolved, + ): + with patch("web_server._post_process_matched_download") as ppm: + ppm.return_value = None + outcome, payload = _process_single_import_file(file_info) + + assert outcome == "ok" + assert payload == "Resolved Title" + + +# --------------------------------------------------------------------------- +# Executor wiring +# --------------------------------------------------------------------------- + + +def test_import_singles_executor_uses_three_workers() -> None: + """Pin the worker count — the user's report (and the maintainer's + acknowledgement) specifically asked for parallelism. Three workers + balance throughput against per-source rate-limit pressure.""" + from web_server import import_singles_executor + + assert import_singles_executor._max_workers == 3 + + +def test_import_singles_executor_threads_are_named_for_diagnostics() -> None: + """Named threads make crash logs and rate-limit diagnostics + immediately attributable to this pool. Without a thread name + prefix, log lines from these workers look identical to the + download workers and post-processing workers.""" + from web_server import import_singles_executor + + assert import_singles_executor._thread_name_prefix == "ImportSingleWorker" + + +# --------------------------------------------------------------------------- +# End-to-end route integration +# --------------------------------------------------------------------------- + + +def test_route_processes_multiple_files_in_parallel(tmp_path) -> None: + """End-to-end: hit the actual /api/import/singles/process route + with multiple files and assert all of them ran. The worker stub + sleeps briefly so a sequential run would be markedly slower than + a 3-worker parallel run; the test pins parallelism by checking + wall-clock duration is well under the sequential cost. + """ + from concurrent.futures import ThreadPoolExecutor + import time as _time + + audio_files = [] + for i in range(6): + f = tmp_path / f"track_{i}.mp3" + f.write_bytes(b"fake audio") + audio_files.append(f) + + files_payload = [ + { + 'full_path': str(f), + 'filename': f.name, + 'title': f"Track {i}", + 'artist': "Test Artist", + } + for i, f in enumerate(audio_files) + ] + + sleep_per_call = 0.3 # 6 files * 0.3s = 1.8s sequential, <0.7s with 3 workers + + def fake_worker(file_info): + _time.sleep(sleep_per_call) + return ("ok", file_info.get('title', '?')) + + from web_server import app as flask_app + flask_app.config['TESTING'] = True + client = flask_app.test_client() + + with patch("web_server._process_single_import_file", side_effect=fake_worker): + start = _time.monotonic() + response = client.post( + "/api/import/singles/process", + json={'files': files_payload}, + ) + duration = _time.monotonic() - start + + assert response.status_code == 200 + payload = response.get_json() + assert payload['success'] is True + assert payload['processed'] == 6 + assert payload['total'] == 6 + assert payload['errors'] == [] + + sequential_cost = sleep_per_call * 6 + # Parallel run with 3 workers should finish in ~2 batches: + # ceil(6 / 3) * 0.3 = 0.6s of sleep + Python overhead. Allow up + # to 2/3 of the sequential cost as the upper bound. + assert duration < sequential_cost * (2 / 3), ( + f"route did not parallelize — took {duration:.2f}s, " + f"sequential would take ~{sequential_cost:.2f}s" + ) + + +def test_route_aggregates_mixed_success_and_error_outcomes(tmp_path) -> None: + """Errors from individual files must not abort the batch; the + final response must list every error and report the success + count separately. Pre-fix, an exception in any single file's + pipeline would propagate up the for-loop's try/except — but + the as_completed loop has its own per-future try/except that's + worth pinning.""" + audio_files = [] + for i in range(4): + f = tmp_path / f"track_{i}.mp3" + f.write_bytes(b"fake") + audio_files.append(f) + + files_payload = [ + {'full_path': str(f), 'filename': f.name, 'title': f"Track {i}", 'artist': 'A'} + for i, f in enumerate(audio_files) + ] + + def mixed_worker(file_info): + # Files 0 and 2 succeed, 1 and 3 fail + idx = int(file_info['filename'].split('_')[1].split('.')[0]) + if idx % 2 == 0: + return ("ok", file_info['title']) + return ("error", f"{file_info['title']}: simulated failure") + + from web_server import app as flask_app + flask_app.config['TESTING'] = True + client = flask_app.test_client() + + with patch("web_server._process_single_import_file", side_effect=mixed_worker): + response = client.post( + "/api/import/singles/process", + json={'files': files_payload}, + ) + + payload = response.get_json() + assert payload['processed'] == 2 + assert payload['total'] == 4 + assert len(payload['errors']) == 2 + assert all('simulated failure' in err for err in payload['errors']) + + +def test_route_recovers_from_worker_crash(tmp_path) -> None: + """If a worker function raises an unhandled exception (shouldn't + happen — the worker wraps its own pipeline call — but defensive), + the route must still finish and report the crash in the errors + list rather than 500-ing the whole batch.""" + audio_files = [tmp_path / f"track_{i}.mp3" for i in range(3)] + for f in audio_files: + f.write_bytes(b"fake") + + files_payload = [ + {'full_path': str(f), 'filename': f.name, 'title': f"T{i}", 'artist': 'A'} + for i, f in enumerate(audio_files) + ] + + call_count = {'n': 0} + + def crashing_worker(file_info): + call_count['n'] += 1 + if call_count['n'] == 2: + raise RuntimeError("worker boom") + return ("ok", file_info['title']) + + from web_server import app as flask_app + flask_app.config['TESTING'] = True + client = flask_app.test_client() + + with patch("web_server._process_single_import_file", side_effect=crashing_worker): + response = client.post( + "/api/import/singles/process", + json={'files': files_payload}, + ) + + assert response.status_code == 200 + payload = response.get_json() + assert payload['success'] is True + assert payload['processed'] == 2 # The two non-crashing calls + assert any('worker crashed' in err for err in payload['errors']) diff --git a/web_server.py b/web_server.py index ace196fc..46493063 100644 --- a/web_server.py +++ b/web_server.py @@ -746,6 +746,15 @@ retag_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="RetagWork # Shared task/batch state now lives in core.runtime_state. missing_download_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="MissingTrackWorker") +# Parallelizes the per-file metadata-lookup + post-processing in +# /api/import/singles/process. Single-file work is dominated by +# Spotify/iTunes/Deezer search round-trips so 3 workers give a near- +# linear speedup on a typical user's network without saturating any +# one provider's rate limit. Each file is independent (unique +# context_key, separate disk path), and the downstream pipeline +# already serializes DB access through its own SQLite locks. +import_singles_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="ImportSingleWorker") + # Automatic Wishlist / Watchlist Processing Flags # Processing state flags (guards/recovery - timers are now managed by AutomationEngine) wishlist_auto_processing = False @@ -2897,6 +2906,7 @@ def _shutdown_runtime_components(): (retag_executor, "retag executor"), (sync_executor, "sync executor"), (missing_download_executor, "missing download executor"), + (import_singles_executor, "import singles executor"), (tidal_discovery_executor, "tidal discovery executor"), (deezer_discovery_executor, "deezer discovery executor"), (spotify_public_discovery_executor, "spotify public discovery executor"), @@ -34101,9 +34111,80 @@ def import_search_tracks(): return jsonify({'success': False, 'error': str(e)}), 500 +def _process_single_import_file(file_info): + """Worker function: validate, resolve metadata, post-process one file. + + Returns ``("ok", title)`` on success, ``("error", message)`` on + failure, or ``("skip", reason)`` for files that need to be reported + but didn't actually run the pipeline. The caller aggregates these. + Designed to be safe to run concurrently from a ThreadPoolExecutor + — each file gets its own UUID context_key, downstream DB writes + serialize via SQLite's busy_timeout, and file-system ops touch + distinct destination paths. + """ + file_path = file_info.get('full_path', '') + if not os.path.isfile(file_path): + return ("error", f"File not found: {file_info.get('filename', '?')}") + + title = file_info.get('title', '') + artist = file_info.get('artist', '') + manual_match = file_info.get('manual_match') + if manual_match is not None and not isinstance(manual_match, dict): + manual_match = None + + manual_match_source = '' + manual_match_id = None + if manual_match: + manual_match_source = str(manual_match.get('source') or '').strip().lower() + manual_match_id = str(manual_match.get('id') or '').strip() + if not manual_match_id or not manual_match_source: + return ("error", f"Malformed manual match for file: {file_info.get('filename', '?')}") + + if not title and not manual_match: + parsed = parse_filename_metadata(file_info.get('filename', '')) + title = parsed.get('title') or os.path.splitext(file_info.get('filename', 'Unknown'))[0] + if not artist: + artist = parsed.get('artist', '') + + from core.imports.resolution import get_single_track_import_context + + try: + resolved = get_single_track_import_context( + title, + artist, + override_id=manual_match_id, + override_source=manual_match_source, + ) + context = normalize_import_context(resolved['context']) + artist_data = get_import_context_artist(context) + track_data = get_import_track_info(context) + final_title = track_data.get('name', title) + final_artist = artist_data.get('name', artist) + + context_key = f"import_single_{uuid.uuid4().hex[:8]}" + _post_process_matched_download(context_key, context, file_path) + logger.info( + "Import single processed: %s by %s (source=%s)", + final_title, + final_artist, + resolved.get('source') or 'local', + ) + return ("ok", final_title) + except Exception as proc_err: + err_msg = f"{title}: {str(proc_err)}" + logger.error(f"Import single processing error: {err_msg}") + return ("error", err_msg) + + @app.route('/api/import/singles/process', methods=['POST']) def import_singles_process(): - """Process individual staging files as singles through the post-processing pipeline.""" + """Process individual staging files as singles through the post-processing pipeline. + + Files are processed in parallel through the + ``import_singles_executor`` (3 workers). Per-file work is dominated + by metadata search round-trips, so parallelizing gives a near- + linear speedup without saturating any one provider's rate limits. + """ try: data = request.get_json() files = data.get('files', []) @@ -34114,63 +34195,30 @@ def import_singles_process(): processed = 0 errors = [] - for file_info in files: - file_path = file_info.get('full_path', '') - if not os.path.isfile(file_path): - errors.append(f"File not found: {file_info.get('filename', '?')}") - continue - - title = file_info.get('title', '') - artist = file_info.get('artist', '') - manual_match = file_info.get('manual_match') - if manual_match is not None and not isinstance(manual_match, dict): - manual_match = None - - manual_match_source = '' - manual_match_id = None - if manual_match: - manual_match_source = str(manual_match.get('source') or '').strip().lower() - manual_match_id = str(manual_match.get('id') or '').strip() - if not manual_match_id or not manual_match_source: - errors.append(f"Malformed manual match for file: {file_info.get('filename', '?')}") - continue - - # Fallback to filename parsing if no metadata - if not title and not manual_match: - parsed = parse_filename_metadata(file_info.get('filename', '')) - title = parsed.get('title') or os.path.splitext(file_info.get('filename', 'Unknown'))[0] - if not artist: - artist = parsed.get('artist', '') - - from core.imports.resolution import get_single_track_import_context - - resolved = get_single_track_import_context( - title, - artist, - override_id=manual_match_id, - override_source=manual_match_source, - ) - context = normalize_import_context(resolved['context']) - artist_data = get_import_context_artist(context) - track_data = get_import_track_info(context) - final_title = track_data.get('name', title) - final_artist = artist_data.get('name', artist) - - context_key = f"import_single_{uuid.uuid4().hex[:8]}" + # Submit all files at once so the executor pulls 3 at a time. + # as_completed yields in finish order; we don't need ordering + # because the caller just wants a count + error list. + future_to_filename = { + import_singles_executor.submit(_process_single_import_file, file_info): + file_info.get('filename', '?') + for file_info in files + } + for future in as_completed(future_to_filename): try: - _post_process_matched_download(context_key, context, file_path) - processed += 1 - logger.info( - "Import single processed: %s by %s (source=%s)", - final_title, - final_artist, - resolved.get('source') or 'local', + outcome, payload = future.result() + except Exception as worker_err: + # Catch-all for anything the worker itself didn't catch + # (shouldn't happen — _process_single_import_file wraps + # its own pipeline call — but defensive). + errors.append( + f"{future_to_filename[future]}: worker crashed: {worker_err}" ) - except Exception as proc_err: - err_msg = f"{title}: {str(proc_err)}" - errors.append(err_msg) - logger.error(f"Import single processing error: {err_msg}") + continue + if outcome == "ok": + processed += 1 + else: + errors.append(payload) add_activity_item("", "Singles Imported", f"{processed}/{len(files)} tracks processed", "Now") From 1aa565a330169bec8b147a1b2b6e1c52da7ee07b Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Thu, 30 Apr 2026 22:58:26 -0700 Subject: [PATCH 2/3] Silence shutdown-time logger errors so CI stderr stays clean Pytest tears down its log file handles before atexit runs. Every "Shutting down ..." line a worker emits while stopping then crashes Python's logger with "I/O operation on closed file" and floods CI stderr with --- Logging error --- traceback blocks. The CI sanity check workflow noticed once tests started importing web_server (the Tidal-auth integration test PR + this parallel-imports PR are the first two test files that boot the full module). Adds a tiny atexit handler that flips ``logging.raiseExceptions = False`` BEFORE the other shutdown handlers run. atexit's LIFO order makes "registered last" mean "runs first", so this fires ahead of cleanup_monitor / _atexit_shutdown / _atexit_save_history and any log calls those make can't bubble the closed-stream traceback. The shutdown messages themselves are best-effort debug breadcrumbs, not data we need to preserve at process exit, so silencing the internal handler errors costs nothing. --- web_server.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/web_server.py b/web_server.py index 46493063..ded6b72e 100644 --- a/web_server.py +++ b/web_server.py @@ -2941,9 +2941,25 @@ def _atexit_shutdown(): except Exception: pass + +def _atexit_silence_shutdown_logger_errors(): + # Pytest tears down log file handles before atexit fires, so every + # "Shutting down ..." line a worker emits while stopping crashes + # Python's logger with "I/O operation on closed file" and floods + # CI stderr. The messages themselves are best-effort debug + # breadcrumbs, not data we need to preserve at process exit. + # Registered last so atexit's LIFO order makes this run FIRST, + # ahead of cleanup_monitor / _atexit_shutdown / _atexit_save_history. + import logging as _logging + _logging.raiseExceptions = False + atexit.register(_atexit_save_history) atexit.register(_atexit_shutdown) atexit.register(cleanup_monitor) +# atexit runs in LIFO order — register the silencer LAST so it runs +# FIRST, before any other shutdown handler emits its "Shutting down" +# log line into a stream pytest already closed. +atexit.register(_atexit_silence_shutdown_logger_errors) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) From ab85c4578585957d514c6b2c17df2a34ba69e9eb Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Fri, 1 May 2026 08:45:17 -0700 Subject: [PATCH 3/3] Restore soulsync logger state between parallel-imports tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Importing web_server fires utils.logging_config.setup_logging at module-init, which clears + re-installs handlers on the shared 'soulsync' logger and pins its level to the user's configured value. That mutation leaks across tests in the same pytest process. This file runs alphabetically before test_library_reorganize_orchestrator, so the leak broke test_watchdog_warns_about_stuck_workers downstream — it relies on caplog capturing soulsync.library_reorganize warnings via root-logger propagation, and the reconfigured logger's new handler chain swallowed those records before they reached caplog (caplog.records came back empty even though pytest's live-log capture clearly showed the warning fired). Adds an autouse fixture that snapshots the soulsync logger's handlers, level, and propagate flag before each test in this file and restores them afterwards. Pollution stays scoped to this file. tests/test_tidal_auth_instructions.py also imports web_server but runs alphabetically AFTER test_library_reorganize_orchestrator so it never tripped this — fix is scoped here, not a project-wide conftest, so we don't change behaviour for unrelated test files. --- tests/test_import_singles_parallel.py | 33 +++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/tests/test_import_singles_parallel.py b/tests/test_import_singles_parallel.py index ff8d1ace..0e72b370 100644 --- a/tests/test_import_singles_parallel.py +++ b/tests/test_import_singles_parallel.py @@ -19,11 +19,44 @@ These tests pin the new behaviour: caller (the test verifies that behaviour through the route). """ +import logging from unittest.mock import patch import pytest +@pytest.fixture(autouse=True) +def _restore_soulsync_logger_state(): + """Snapshot the ``soulsync`` logger config before this file's tests + run and restore it afterwards. + + Importing ``web_server`` calls ``utils.logging_config.setup_logging`` + at module-init time, which clears + re-installs handlers on the + ``soulsync`` logger and pins its level to whatever the user's + config said. That mutation leaks across tests in the same pytest + process and broke + ``test_library_reorganize_orchestrator::test_watchdog_warns_about_stuck_workers`` + that runs later alphabetically and relies on caplog capturing + ``soulsync.library_reorganize`` warnings via root-logger + propagation. + + Without this fixture, my file ran first alphabetically, mutated + the global soulsync logger, and the watchdog test downstream + saw ``caplog.records == []``. Snapshot + restore keeps the + pollution scoped to this file's tests only. + """ + soulsync_logger = logging.getLogger("soulsync") + saved_handlers = list(soulsync_logger.handlers) + saved_level = soulsync_logger.level + saved_propagate = soulsync_logger.propagate + try: + yield + finally: + soulsync_logger.handlers = saved_handlers + soulsync_logger.setLevel(saved_level) + soulsync_logger.propagate = saved_propagate + + # --------------------------------------------------------------------------- # Worker contract # ---------------------------------------------------------------------------