fix(post-processing): prevent double-claim race that fails imported tracks

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 <noreply@anthropic.com>

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

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.

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()

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

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