danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Parallelize singles-import processing with a 3-worker executor

Browse source 
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.
This commit is contained in:
Broque Thomas 2026-04-30 22:41:04 -07:00
parent 7f191be7af
commit f339211654
2 changed files with 405 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

302
tests/test_import_singles_parallel.py Normal file
View file

@ -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'])

158
web_server.py
View file

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