05b36704c3
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>
117 lines
4 KiB
Python
117 lines
4 KiB
Python
"""Shared runtime state and tiny helpers for the app."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from functools import wraps
|
|
from typing import Any, Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
matched_context_lock = threading.Lock()
|
|
matched_downloads_context: Dict[str, Dict[str, Any]] = {}
|
|
tasks_lock = threading.Lock()
|
|
download_tasks: Dict[str, Dict[str, Any]] = {}
|
|
download_batches: Dict[str, Dict[str, Any]] = {}
|
|
batch_locks: Dict[str, threading.Lock] = {}
|
|
processed_download_ids = set()
|
|
post_process_locks: Dict[str, threading.Lock] = {}
|
|
post_process_locks_lock = threading.Lock()
|
|
|
|
activity_feed = []
|
|
activity_feed_lock = threading.Lock()
|
|
_activity_toast_emitter = None
|
|
|
|
|
|
def caller_must_hold_tasks_lock(func):
|
|
"""Best-effort guard for helpers that mutate download_tasks in place."""
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
if not tasks_lock.locked():
|
|
raise RuntimeError(f"{func.__name__}() requires tasks_lock to be held by the caller")
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def set_activity_toast_emitter(emitter) -> None:
|
|
"""Set the WebSocket-style emitter used by add_activity_item."""
|
|
global _activity_toast_emitter
|
|
_activity_toast_emitter = emitter
|
|
|
|
|
|
def add_activity_item(icon, title, subtitle, time_ago="Now", show_toast=True):
|
|
"""Append an activity item and emit a toast if an emitter is configured."""
|
|
activity_item = {
|
|
"icon": icon,
|
|
"title": title,
|
|
"subtitle": subtitle,
|
|
"time": time_ago,
|
|
"timestamp": time.time(),
|
|
"show_toast": show_toast,
|
|
}
|
|
with activity_feed_lock:
|
|
activity_feed.append(activity_item)
|
|
if len(activity_feed) > 20:
|
|
activity_feed.pop(0)
|
|
|
|
if show_toast and _activity_toast_emitter is not None:
|
|
try:
|
|
_activity_toast_emitter("dashboard:toast", activity_item)
|
|
except Exception as e:
|
|
logger.debug("emit activity toast failed: %s", e)
|
|
|
|
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.
|
|
|
|
Callers must already hold `tasks_lock`.
|
|
"""
|
|
task = download_tasks.get(task_id)
|
|
if not task:
|
|
return False
|
|
|
|
task["status"] = "completed"
|
|
task["stream_processed"] = True
|
|
task["status_change_time"] = time.time()
|
|
if track_info is not None:
|
|
task["track_info"] = track_info
|
|
return True
|