From 05b36704c30c2bc87e5ef75114a4336d9e88e5e7 Mon Sep 17 00:00:00 2001 From: dev Date: Wed, 24 Jun 2026 17:39:19 +0200 Subject: [PATCH 1/4] fix(post-processing): prevent double-claim race that fails imported tracks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two subsystems post-process the same completed transfer: the browser-poll status endpoint (web_server) and the background download monitor. Both watch the same slskd/streaming transfers and each launches the verification pipeline. When one path quarantines + requeues the next-best candidate (clearing username/filename, status -> 'searching'), the monitor's already-submitted run_post_processing_worker then runs, finds no source info, and falsely marks the task 'failed' ("missing file or source information") — clobbering the in-flight retry while a parallel attempt imports the song. Fix: a single atomic claim (downloading/queued -> post_processing under tasks_lock) so exactly one path processes each download. - runtime_state: new claim_for_post_processing() helper - post_processing: race guard — worker bails (no fail/notify) if the task is no longer 'post_processing' when it runs - web_server: both poll paths (Soulseek + streaming) claim before launching; claim is released on thread-launch failure Co-Authored-By: Claude Opus 4.8 --- core/downloads/post_processing.py | 15 +++++++ core/runtime_state.py | 33 +++++++++++++++ .../test_downloads_post_processing.py | 25 ++++++++++++ tests/test_runtime_state.py | 31 ++++++++++++++ web_server.py | 40 +++++++++++++++++++ 5 files changed, 144 insertions(+) diff --git a/core/downloads/post_processing.py b/core/downloads/post_processing.py index 2c17e90f..5dbbae97 100644 --- a/core/downloads/post_processing.py +++ b/core/downloads/post_processing.py @@ -171,6 +171,21 @@ def run_post_processing_worker(task_id: str, batch_id: str, deps: PostProcessDep logger.info(f"[Post-Processing] Task {task_id} already completed by stream processor, skipping verification") return + # RACE GUARD: the monitor sets status -> 'post_processing' immediately + # before submitting this worker. If the status is now anything else, the + # browser-poll post-processor already took ownership of this task — e.g. + # it quarantined the file and requeued the next-best candidate (status + # -> 'searching', source identity cleared). Bail WITHOUT marking failed + # or notifying batch completion: otherwise we clobber that in-flight + # retry with a bogus "missing file or source information" failure while a + # parallel attempt is importing the song. + if task['status'] != 'post_processing': + logger.info( + f"[Post-Processing] Task {task_id} no longer in 'post_processing' " + f"(now '{task['status']}') — another path took over, skipping" + ) + return + # Extract file information for verification track_info = task.get('track_info', {}) task_filename = task.get('filename') or track_info.get('filename') diff --git a/core/runtime_state.py b/core/runtime_state.py index dfa11798..f4528366 100644 --- a/core/runtime_state.py +++ b/core/runtime_state.py @@ -66,6 +66,39 @@ def add_activity_item(icon, title, subtitle, time_ago="Now", show_toast=True): return activity_item +def claim_for_post_processing(task_id: str) -> bool: + """Atomically claim a download task for post-processing. + + The browser-poll status endpoint AND the background download monitor both + watch the same slskd/streaming transfers and each tries to post-process a + completed file. Without a single claim, BOTH run the verification pipeline + on the same download — double imports, and a nasty race where one path + quarantines + requeues the next-best candidate (clearing the source identity + and resetting status to ``searching``) while the other, mid-flight, then + reports a bogus "missing file or source information" failure that clobbers + the in-flight retry. + + The claim is the ``downloading``/``queued`` -> ``post_processing`` status + transition, done under ``tasks_lock``. Exactly one caller wins: + + - Returns ``True`` (and flips the status) for the caller that claimed it. + - Returns ``False`` if the task is gone, already ``post_processing`` (owned + by the other path), requeued (``searching``), or terminal — the caller + must then NOT process the file. + + Acquires ``tasks_lock`` itself, so callers must NOT already hold it. + """ + with tasks_lock: + task = download_tasks.get(task_id) + if not task: + return False + if task.get("status") in ("downloading", "queued"): + task["status"] = "post_processing" + task["status_change_time"] = time.time() + return True + return False + + @caller_must_hold_tasks_lock def mark_task_completed(task_id: str, track_info: Optional[Dict[str, Any]] = None) -> bool: """Mark a download task as completed. diff --git a/tests/downloads/test_downloads_post_processing.py b/tests/downloads/test_downloads_post_processing.py index 455b2237..701c6e2b 100644 --- a/tests/downloads/test_downloads_post_processing.py +++ b/tests/downloads/test_downloads_post_processing.py @@ -117,6 +117,31 @@ def test_stream_processed_task_returns_early(): assert rec.calls == [] +def test_requeued_task_bails_without_marking_failed(): + """RACE GUARD: the monitor sets status -> 'post_processing' and submits this + worker. If, before the worker runs, the browser-poll post-processor + quarantines the file and requeues the next-best candidate (status -> + 'searching', username/filename cleared), this worker must bail WITHOUT + marking failed or notifying batch completion. Otherwise it clobbers the + in-flight retry with a false 'missing file or source information' failure + while a parallel attempt imports the song.""" + download_tasks['t1'] = {'status': 'searching', 'track_info': {}} + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'searching' # untouched + assert 'error_message' not in download_tasks['t1'] + assert not any(c[0] == 'on_complete' for c in rec.calls) + + +def test_queued_task_bails_without_marking_failed(): + """Same race guard for a task another path reset to 'queued'.""" + download_tasks['t1'] = {'status': 'queued', 'track_info': {}} + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'queued' + assert not any(c[0] == 'on_complete' for c in rec.calls) + + def test_missing_filename_marks_failed_and_calls_on_complete(): download_tasks['t1'] = {'status': 'post_processing', 'username': 'u1', 'track_info': {}} deps, rec = _build_deps() diff --git a/tests/test_runtime_state.py b/tests/test_runtime_state.py index 87fb779f..f66a10f2 100644 --- a/tests/test_runtime_state.py +++ b/tests/test_runtime_state.py @@ -30,3 +30,34 @@ def test_mark_task_completed_succeeds_when_lock_held(): finally: runtime_state.download_tasks.clear() runtime_state.download_tasks.update(original_tasks) + + +@pytest.fixture +def _isolated_tasks(): + original_tasks = dict(runtime_state.download_tasks) + runtime_state.download_tasks.clear() + yield runtime_state.download_tasks + runtime_state.download_tasks.clear() + runtime_state.download_tasks.update(original_tasks) + + +@pytest.mark.parametrize("start_status", ["downloading", "queued"]) +def test_claim_for_post_processing_wins_from_active_status(_isolated_tasks, start_status): + _isolated_tasks["t1"] = {"status": start_status} + assert runtime_state.claim_for_post_processing("t1") is True + assert _isolated_tasks["t1"]["status"] == "post_processing" + + +@pytest.mark.parametrize("start_status", ["post_processing", "searching", "completed", "failed", "cancelled"]) +def test_claim_for_post_processing_loses_when_already_owned(_isolated_tasks, start_status): + """A task already being post-processed by the monitor (post_processing), + requeued by a quarantine retry (searching), or already terminal must NOT be + re-claimed — that is the double-processing race that produced the bogus + 'missing file or source information' failure.""" + _isolated_tasks["t1"] = {"status": start_status} + assert runtime_state.claim_for_post_processing("t1") is False + assert _isolated_tasks["t1"]["status"] == start_status # untouched + + +def test_claim_for_post_processing_missing_task_returns_false(_isolated_tasks): + assert runtime_state.claim_for_post_processing("absent") is False diff --git a/web_server.py b/web_server.py index 0c81f2f3..5d14f380 100644 --- a/web_server.py +++ b/web_server.py @@ -192,6 +192,7 @@ from core.runtime_state import ( activity_feed, activity_feed_lock, add_activity_item, + claim_for_post_processing, download_batches, download_tasks, matched_context_lock, @@ -6796,6 +6797,19 @@ def get_download_status(): _pp_task_id = context.get('task_id') _pp_batch_id = context.get('batch_id') if _pp_task_id and _pp_batch_id: + # Atomic claim: the download monitor ALSO watches slskd + # transfers and submits its own post-processing worker for + # this task. Losing the claim means the monitor (or a + # parallel poll) already owns it — skip to avoid a double + # import and the quarantine-requeue race that produced the + # bogus "missing file or source information" failure. + if not claim_for_post_processing(_pp_task_id): + logger.info( + f"Task {_pp_task_id} already claimed for post-processing " + f"by another path — skipping duplicate for {context_key}" + ) + processed_download_ids.add(context_key) + continue _pp_target = _post_process_matched_download_with_verification _pp_args = (context_key, context, found_path, _pp_task_id, _pp_batch_id) else: @@ -6817,6 +6831,15 @@ def get_download_status(): logger.error(f"Error starting post-processing thread for {context_key}: {e}") # Don't add to processed set if thread failed to start logger.warning(f"Will retry {context_key} on next check") + # Release the post-processing claim so a later poll (or the + # monitor) can retry — otherwise the task is stuck in + # 'post_processing' with no worker running. + _failed_task_id = context.get('task_id') + if _failed_task_id: + with tasks_lock: + _ft = download_tasks.get(_failed_task_id) + if _ft and _ft.get('status') == 'post_processing': + _ft['status'] = 'downloading' # Start a single thread to manage the launching of all processing threads processing_thread = threading.Thread(target=process_completed_downloads) @@ -6874,6 +6897,16 @@ def get_download_status(): _st_task_id = _ctx.get('task_id') _st_batch_id = _ctx.get('batch_id') if _st_task_id and _st_batch_id: + # Atomic claim — the download monitor watches + # these (non-soulseek) transfers too and would + # otherwise double-process this same task. + if not claim_for_post_processing(_st_task_id): + logger.info( + f"[{_label}] Task {_st_task_id} already claimed " + f"for post-processing — skipping duplicate for {_ctx_key}" + ) + processed_download_ids.add(_ctx_key) + return _st_target = _post_process_matched_download_with_verification _st_args = (_ctx_key, _ctx, _path, _st_task_id, _st_batch_id) else: @@ -6886,6 +6919,13 @@ def get_download_status(): logger.info(f"[{_label}] Marked as processed: {_ctx_key}") except Exception as e: logger.error(f"[{_label}] Error starting post-processing thread for {_ctx_key}: {e}") + # Release the claim so a later poll / the monitor retries. + _st_failed_id = _ctx.get('task_id') + if _st_failed_id: + with tasks_lock: + _stf = download_tasks.get(_st_failed_id) + if _stf and _stf.get('status') == 'post_processing': + _stf['status'] = 'downloading' processing_thread = threading.Thread(target=process_streaming_download) processing_thread.daemon = True From ab05508acc051d9e92acc707456a6e48c945379c Mon Sep 17 00:00:00 2001 From: dev Date: Thu, 25 Jun 2026 22:00:57 +0200 Subject: [PATCH 2/4] feat(quality): rank-based candidate ordering toggle for priority mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an opt-in `rank_candidates_by_quality` profile flag. When on, the priority-mode download walk orders candidates by the ranked-target quality (confidence/speed only break ties) instead of confidence-first. Default off keeps the byte-for-byte old behaviour, so existing installs are unaffected. Best-quality search mode is always quality-first regardless of the flag; the toggle only affects priority mode. Search-time source selection is unchanged — nothing is skipped, so a track can never go missing, only the order in which copies are tried changes. The version-mismatch force-import follows automatically: it accepts the first-tried (= best-ordered) quarantined candidate, which is the highest-quality one once the walk is quality-first. No change to its selection logic needed. - core/quality/selection.py: load_rank_candidates_by_quality() (fail-closed). - core/downloads/task_worker.py: _best_quality_ordering -> _candidate_ordering; quality-first when best_quality mode OR the toggle is on. - database/music_database.py: default profile carries the flag (False). - web_server.py: flag is preserved globally across preset apply/reset, like search_mode. - core/imports/version_mismatch_fallback.py: comment clarified (no behaviour change). Tests (TDD): load_rank_candidates_by_quality default/enabled/disabled/error; _candidate_ordering across all mode+toggle combinations + fail-closed. Co-Authored-By: Claude Opus 4.8 --- core/downloads/task_worker.py | 35 +++++++++----- core/imports/version_mismatch_fallback.py | 9 ++-- core/quality/selection.py | 20 ++++++++ database/music_database.py | 1 + .../downloads/test_worker_quality_ordering.py | 47 +++++++++++++++++++ tests/quality/test_search_mode.py | 29 ++++++++++++ web_server.py | 9 +++- 7 files changed, 134 insertions(+), 16 deletions(-) create mode 100644 tests/downloads/test_worker_quality_ordering.py diff --git a/core/downloads/task_worker.py b/core/downloads/task_worker.py index 1b7b2bc1..f615de47 100644 --- a/core/downloads/task_worker.py +++ b/core/downloads/task_worker.py @@ -54,21 +54,34 @@ def _cand_user_file(candidate): return getattr(candidate, 'username', None), getattr(candidate, 'filename', None) -def _best_quality_ordering(): - """Return ``(quality_first, targets)`` for the active search mode. +def _candidate_ordering(): + """Return ``(quality_first, targets)`` for the active search mode + toggle. - In best-quality mode the candidate walk is ordered by the user's profile - quality rank (best→worst) instead of confidence-first. Fails closed to - priority-mode ordering on any error so a profile/DB hiccup never blocks a - download. See docs/superpowers/specs/2026-06-14-best-quality-search-mode-design.md. + The candidate walk is ordered by the user's profile quality rank + (best→worst) instead of confidence-first when EITHER: + - best-quality search mode is active (always quality-first), OR + - priority mode and the ``rank_candidates_by_quality`` toggle is on + (opt-in; default off keeps the byte-for-byte confidence-first walk). + + Quality-first ordering also makes the version-mismatch force-import pick + the highest-quality candidate, because that fallback accepts the + first-tried (= best-ordered) quarantined entry. + + Fails closed to confidence-first ordering on any error so a profile/DB + hiccup never blocks a download. See + docs/superpowers/specs/2026-06-14-best-quality-search-mode-design.md. """ try: - from core.quality.selection import load_search_mode, load_profile_targets - if load_search_mode() == 'best_quality': + from core.quality.selection import ( + load_search_mode, + load_profile_targets, + load_rank_candidates_by_quality, + ) + if load_search_mode() == 'best_quality' or load_rank_candidates_by_quality(): targets, _ = load_profile_targets() return True, targets except Exception as exc: - logger.debug("[Modal Worker] best-quality ordering unavailable: %s", exc) + logger.debug("[Modal Worker] quality ordering unavailable: %s", exc) return False, None @@ -109,7 +122,7 @@ def _try_cached_candidates(task_id, batch_id, track, deps): f"[Modal Worker] Quarantine retry: trying {len(remaining)} cached " f"candidate(s) before re-searching (task {task_id})" ) - _qf, _qt = _best_quality_ordering() + _qf, _qt = _candidate_ordering() return deps.attempt_download_with_candidates( task_id, remaining, track, batch_id, quality_first=_qf, quality_targets=_qt, ) @@ -393,7 +406,7 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke # Best-quality search mode: the orchestrator already pooled candidates # across every source for each query, so order the candidate walk by the # user's profile quality rank (best→worst). Computed once per task. - _best_quality, _quality_targets = _best_quality_ordering() + _best_quality, _quality_targets = _candidate_ordering() # 2. Sequential Query Search (matches GUI's start_search_worker_parallel logic) search_diagnostics = [] # Track what happened per query for detailed error messages diff --git a/core/imports/version_mismatch_fallback.py b/core/imports/version_mismatch_fallback.py index 3e0ef92d..2bf37388 100644 --- a/core/imports/version_mismatch_fallback.py +++ b/core/imports/version_mismatch_fallback.py @@ -103,9 +103,12 @@ def select_version_mismatch_fallback( # don't guess which the user wants. return None - # First tried = oldest = highest-confidence (the retry walks candidates - # best-first). The id is a "_