Release 0.20.0

app/api/events.py

@@ -8,6 +8,7 @@
 from fastapi.responses import StreamingResponse
 
 from app.core.config import JOB_ID_RE
+from app.core.models import FINISHED_STATUSES
 from app.core.registry import get as registry_get
 
 router = APIRouter(tags=["events"])
@@ -108,9 +109,9 @@ async def stream() -> AsyncIterator[str]:
                     yield f"data: {json.dumps(snapshot)}\n\n"
                     last_v = v
                     keepalive_at = 0
-                    if snapshot["status"] in ("done", "error", "cancelled"):
+                    if snapshot["status"] in FINISHED_STATUSES:
                         return
-                elif job.status in ("done", "error", "cancelled"):
+                elif job.status in FINISHED_STATUSES:
                     # Already-terminal with no pending change (e.g. the job was
                     # done before this connection opened) -- close promptly
                     # instead of idling on int-compares until the SSE cap.

app/api/jobs.py

@@ -35,15 +35,17 @@
     TIMEOUT_WORK_BACKFILL,
     ffprobe_executable,
 )
-from app.core.models import Job, JobCancelled, _set
+from app.core.models import FINISHED_STATUSES, Job, JobCancelled, _set, settle_cancelled
 from app.core.registry import all_jobs as registry_all_jobs
 from app.core.registry import get as registry_get
 from app.core.registry import get_proc as registry_get_proc
+from app.core.registry import is_upload
 from app.core.registry import mark_deleted as registry_mark_deleted
 from app.core.registry import pending_count as registry_pending_count
 from app.core.registry import persist as registry_persist
 from app.core.registry import register_if_capacity as registry_register_if_capacity
 from app.core.registry import remove as registry_remove
+from app.core.registry import requeue_if_capacity as registry_requeue_if_capacity
 from app.core.registry import set_favorite as registry_set_favorite
 from app.core.registry import set_trashed as registry_set_trashed
 from app.core.settings import (
@@ -56,7 +58,7 @@
 from app.pipeline import discogs, jobqueue
 from app.pipeline.artist_lookup import tagged_artist_name
 from app.pipeline.audio_tags import probe_tags
-from app.pipeline.collect import merge_stem_peaks, presence_for_split
+from app.pipeline.collect import merge_stem_peaks, presence_for_split, remove_job_dir
 from app.pipeline.download import InvalidYouTubeURL, fetch_audio_tags, validate_youtube_url
 from app.pipeline.errors import classify_failure
 from app.pipeline.identify import can_identify_title, identify_and_find_band, release_source
@@ -168,23 +170,9 @@ def _rmtree_job(job_id: str) -> bool:
     The outcome used to be swallowed, so delete_job dropped the registry entry
     whether or not anything was actually deleted -- and restore() then adopted
     the surviving directory on the next start, which is how deleted songs came
-    back (#521).
-
-    Retried once: on macOS the common failure is Finder or Spotlight creating
-    a .DS_Store between rmtree's scan and its final rmdir, which leaves
-    "Directory not empty" on a directory that is about to be empty again."""
-    job_dir = JOBS_DIR / job_id
-    for attempt in (1, 2):
-        if not job_dir.is_dir():
-            return True
-        try:
-            shutil.rmtree(job_dir)
-            return True
-        except Exception:
-            logger.warning(
-                "failed to remove job dir %s (attempt %d)", job_dir, attempt, exc_info=True
-            )
-    return not job_dir.is_dir()
+    back (#521). The retry lives in remove_job_dir, shared with the runner's
+    cancel path (#749)."""
+    return remove_job_dir(JOBS_DIR / job_id)
 
 
 def _job_files_missing(job: Job) -> bool:
@@ -403,12 +391,24 @@ def list_jobs(trashed: Literal["exclude", "include", "only"] = "exclude") -> lis
 
 @router.post("/{job_id}/trash")
 def trash_job(job_id: str) -> dict:
-    """Put a job in the Trash. Reversible, and nothing on disk is touched."""
+    """Put a job in the Trash. Reversible, and nothing on disk is touched:
+    only emptying the Trash (DELETE) removes files.
+
+    A job still waiting or running is stopped as well (#748). Trashing used to
+    leave it going, hidden: the queue kept working on a song the user had
+    thrown away, and a stuck one held the queue up with no way to clear it. It
+    settles as "stopped", its files kept, so Restore brings it back and Extract
+    can run it again. Done here rather than in each client so the phone and the
+    desktop behave alike.
+    """
     if not JOB_ID_RE.match(job_id):
         raise HTTPException(status_code=404, detail="job not found")
     job = registry_set_trashed(job_id, True)
     if job is None:
         raise HTTPException(status_code=404, detail="job not found")
+    if job.status not in FINISHED_STATUSES:
+        job.stop_requested = True
+        _cancel_pipeline(job)
     registry_persist(JOBS_DIR)
     return {"job_id": job.id, "trashed_at": job.trashed_at}
 
@@ -425,6 +425,35 @@ def restore_job(job_id: str) -> dict:
     return {"job_id": job.id, "trashed_at": job.trashed_at}
 
 
+@router.post("/{job_id}/extract")
+def extract_stopped_job(job_id: str) -> dict:
+    """Run a stopped job again from the start, from the source it kept (#748).
+
+    The only way a job stopped by the Trash goes back in the queue. Restore
+    does not do it: taking a song out of the Trash is not a request to spend
+    minutes of GPU on it. It has to be out of the Trash first, so a song the
+    user threw away is never extracted by a stray request.
+    """
+    if not JOB_ID_RE.match(job_id):
+        raise HTTPException(status_code=404, detail="job not found")
+    job = registry_get(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="job not found")
+    if job.status != "stopped":
+        raise HTTPException(status_code=409, detail="only a stopped job can be extracted again")
+    if job.trashed_at is not None:
+        raise HTTPException(status_code=409, detail="restore it from the Trash first")
+    limit = MAX_PENDING_UPLOAD_JOBS if is_upload(job) else MAX_PENDING_URL_JOBS
+    if not registry_requeue_if_capacity(job, JOBS_DIR / job_id, limit):
+        raise HTTPException(
+            status_code=503,
+            detail=_UPLOAD_QUEUE_FULL_DETAIL if is_upload(job) else _URL_QUEUE_FULL_DETAIL,
+        )
+    registry_persist(JOBS_DIR)
+    jobqueue.enqueue(job_id)
+    return job.to_state()
+
+
 class FavoriteBody(BaseModel):
     """Strict, so a string such as "false" is refused rather than read as true."""
 
@@ -459,7 +488,7 @@ def cancel_job(job_id: str) -> dict:
     job = registry_get(job_id)
     if job is None:
         raise HTTPException(status_code=404, detail="job not found")
-    if job.status in ("done", "error", "cancelled"):
+    if job.status in FINISHED_STATUSES:
         # A vocal split only ever runs on a done job, so this early return made
         # it uncancellable by construction: the flag was never even set, while
         # the split held _pipeline_lock and stalled the whole import queue for
@@ -478,17 +507,30 @@ def cancel_job(job_id: str) -> dict:
             if proc is not None and proc.poll() is None:
                 proc.terminate()
         return job.to_state()
+    _cancel_pipeline(job)
+    return job.to_state()
+
+
+def _cancel_pipeline(job: Job) -> None:
+    """Stop a job that is waiting or running. Shared by cancel and trash (#748).
+
+    The caller has checked the job is not finished. A running job is only
+    asked to stop here: the pipeline sees the flag, or its process dies, and
+    the runner then settles it: cancelled with its directory removed, or stopped
+    with its files kept when the stop came from the Trash.
+    """
+    job_id = job.id
     job.cancel_requested = True
 
     # Still waiting: the worker will never pick it up, so finalise it here.
     # Previously a queued job only honoured cancel once its turn arrived, kept
     # occupying a capacity slot until then, and a queued upload held its source
     # file (up to 400 MB) for the whole wait.
     if jobqueue.discard(job_id):
-        _set(job, status="cancelled", stage="Cancelled")
-        jobqueue.cleanup_job_dir(job_id)
+        if settle_cancelled(job):
+            jobqueue.cleanup_job_dir(job_id)
         registry_persist(JOBS_DIR)
-        return job.to_state()
+        return
 
     # Only the running job owns the shared demucs worker. Terminating on any
     # other id would kill someone else's separation if a stale set_proc entry
@@ -497,7 +539,6 @@ def cancel_job(job_id: str) -> dict:
         proc = registry_get_proc(job_id)
         if proc is not None and proc.poll() is None:
             proc.terminate()
-    return job.to_state()
 
 
 def _write_vocal_split_error(stems_dir: Path, cause: str, tail: list[str]) -> None:
@@ -1802,7 +1843,7 @@ def delete_job(job_id: str) -> dict[str, str]:
     job = registry_get(job_id)
     if job is None:
         raise HTTPException(status_code=404, detail="job not found")
-    if job.status not in ("done", "error", "cancelled"):
+    if job.status not in FINISHED_STATUSES:
         raise HTTPException(status_code=409, detail="job is still running")
     # A timing of its lyrics in flight stops rather than write into a
     # directory being removed (it also checks the job is still registered).

app/core/models.py

@@ -15,9 +15,22 @@ class JobCancelled(Exception):
 
 
 JobStatus = Literal[
-    "queued", "downloading", "analyzing", "separating", "processing", "done", "error", "cancelled"
+    "queued",
+    "downloading",
+    "analyzing",
+    "separating",
+    "processing",
+    "done",
+    "error",
+    "cancelled",
+    "stopped",
 ]
 
+# Statuses no pipeline will move on from. "stopped" is a job halted because the
+# user put it in the Trash (#748): unlike "cancelled", its files stay on disk,
+# because only emptying the Trash deletes anything.
+FINISHED_STATUSES = frozenset(("done", "error", "cancelled", "stopped"))
+
 # A Wikidata item id. Anything else is not a band this app found.
 _WIKIDATA_ID_RE = re.compile(r"^Q\d{1,12}$")
 # A band's name, as kept on the job. Longer than this is not a name.
@@ -195,6 +208,20 @@ def _set(job: Job, **fields: object) -> None:
     job.version += 1
 
 
+def settle_cancelled(job: Job) -> bool:
+    """Mark a job whose cancel took effect. Returns whether its files go too.
+
+    A plain cancel ends "cancelled" and its directory is removed. A stop that
+    came from the Trash ends "stopped" with the directory kept (#748): trashing
+    is reversible, and only emptying the Trash deletes files, the rule the
+    finished tracks in the Trash have always followed."""
+    if job.stop_requested:
+        _set(job, status="stopped", stage="Stopped")
+        return False
+    _set(job, status="cancelled", stage="Cancelled")
+    return True
+
+
 @dataclass
 class Job:
     id: str
@@ -319,6 +346,10 @@ class Job:
     # Set by POST /api/jobs/{id}/cancel; consumed by pipeline stages.
     # Not surfaced via to_state() -- it's internal control state.
     cancel_requested: bool = False
+    # Set alongside cancel_requested when the stop comes from the Trash (#748).
+    # The job then settles as "stopped" with its files kept, not "cancelled"
+    # with its directory removed. Internal, like cancel_requested.
+    stop_requested: bool = False
     # Bumped by _set() on every field write (#289). Internal dirty-flag /
     # tear-detection state for the SSE stream -- not surfaced via to_state()
     # or persisted, same as cancel_requested.
@@ -412,9 +443,12 @@ def from_record(cls, data: dict[str, Any]) -> Job:
         job.work = clean_work(job.work)
         job.audio_tags = clean_audio_tags(job.audio_tags)
         job.cancel_requested = False
+        job.stop_requested = False
         return job
 
 
 _JOB_FIELDS = frozenset(
-    f.name for f in dataclasses.fields(Job) if f.name not in ("cancel_requested", "version")
+    f.name
+    for f in dataclasses.fields(Job)
+    if f.name not in ("cancel_requested", "stop_requested", "version")
 )

app/core/registry.py

@@ -29,7 +29,10 @@
 # handed back to the queue -- see restore(). _TERMINAL stays narrow because
 # restore() and _recover_done_job() both mean "finished successfully" by it.
 _RESUMABLE = {"queued", "processing", "downloading", "analyzing", "separating"}
-_PERSISTED = _TERMINAL | _RESUMABLE
+# Halted from the Trash with its files kept (#748). Persisted so a restart
+# keeps it in the Trash, and never resumed: only Extract queues it again.
+_STOPPED = {"stopped"}
+_PERSISTED = _TERMINAL | _RESUMABLE | _STOPPED
 
 # One resume only. A job that reliably kills the process (a demucs OOM taking
 # the backend down with it) would otherwise be re-queued on every start,
@@ -65,6 +68,23 @@ def pending_count(*, uploads: bool) -> int:
         return sum(1 for j in _jobs.values() if j.status == "queued" and is_upload(j) == uploads)
 
 
+def requeue_if_capacity(job: Job, job_dir: Path, max_pending: int) -> bool:
+    """Reset a stopped job and count it as waiting, if its kind has room.
+
+    The same per-kind limit as a new import, checked and applied under one lock
+    so two Extract presses cannot both squeeze into the last slot (#748)."""
+    uploads = is_upload(job)
+    with _lock:
+        pending = sum(1 for j in _jobs.values() if j.status == "queued" and is_upload(j) == uploads)
+        if pending >= max_pending:
+            return False
+        # Claims the slot. The file work in reset_for_rerun stays outside the
+        # lock; the job is not in the queue until the caller enqueues it.
+        job.status = "queued"
+    reset_for_rerun(job, job_dir)
+    return True
+
+
 def register_if_capacity(job: Job, max_pending: int) -> bool:
     """Atomically check pending count and register if under capacity.
     Returns True if registered, False if the queue is full.
@@ -218,7 +238,9 @@ def restore(jobs_dir: Path) -> None:
                 job = Job.from_record(record)
                 if not JOB_ID_RE.match(job.id):
                     continue
-                if job.status in _TERMINAL and job.title:
+                # A stopped job may have no title yet: one trashed before its
+                # download finished still has files and a place in the Trash.
+                if (job.status in _TERMINAL and job.title) or job.status in _STOPPED:
                     to_add[job.id] = job
                 elif job.status in _RESUMABLE:
                     recovered = _resume_or_recover(job, jobs_dir / job.id)
@@ -285,26 +307,52 @@ def _resume_or_recover(job: Job, job_dir: Path) -> Job | None:
     library entry. Otherwise the job goes back in the queue from the top, after
     the partial demucs output is cleared so collect() cannot mistake it for
     results. A job that has already burned its resume is failed loudly rather
-    than retried forever."""
+    than retried forever.
+
+    A job the user trashed before the restart is not resumed (#748). It came
+    back as a Paused row in the queue, offered to Start a song that was in the
+    Trash. It is stopped instead, the same end a trash before the restart now
+    has, and its files stay until the Trash is emptied."""
     recovered = _recover_done_job(job_dir)
     if recovered is not None:
+        # Finished after all: keep it where the user put it, or a song they
+        # binned reappears in the library.
+        recovered.trashed_at = job.trashed_at
         return recovered
 
+    if job.trashed_at is not None:
+        reset_for_rerun(job, job_dir)
+        job.status = "stopped"
+        job.stage_message = "Stopped"
+        return job
+
     job.resume_attempts += 1
     if job.resume_attempts > _MAX_RESUME_ATTEMPTS:
         job.status = "error"
         job.stage_message = "Error: Interrupted"
         job.error = "Interrupted by a restart twice. Import it again to retry."
         return job
 
+    reset_for_rerun(job, job_dir)
+    return job
+
+
+def reset_for_rerun(job: Job, job_dir: Path) -> None:
+    """Put a job back to the start of the pipeline, its source kept.
+
+    The partial demucs output is cleared so collect() cannot mistake it for
+    results. Shared by a resume after a restart and by extracting a stopped
+    job again (#748). Bumps version, so a stream watching the job sees it."""
     shutil.rmtree(job_dir / DEMUCS_MODEL, ignore_errors=True)
     job.status = "queued"
     job.stage_message = "Queued"
     job.progress = 0.0
     job.cancel_requested = False
+    job.stop_requested = False
+    job.error = None
     job.stems = []
     job.mix_url = None
-    return job
+    job.version += 1
 
 
 def take_pending_resume() -> list[str]: