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] 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")