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