diff --git a/app/api/events.py b/app/api/events.py index dedb0b84..24f683f0 100644 --- a/app/api/events.py +++ b/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. diff --git a/app/api/jobs.py b/app/api/jobs.py index eecca686..af132f5b 100644 --- a/app/api/jobs.py +++ b/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,6 +507,19 @@ 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. @@ -485,10 +527,10 @@ def cancel_job(job_id: str) -> dict: # 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). diff --git a/app/core/models.py b/app/core/models.py index e51c84d2..4e0a1b85 100644 --- a/app/core/models.py +++ b/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") ) diff --git a/app/core/registry.py b/app/core/registry.py index 9b16793f..69cffe21 100644 --- a/app/core/registry.py +++ b/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,11 +307,25 @@ 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" @@ -297,14 +333,26 @@ def _resume_or_recover(job: Job, job_dir: Path) -> Job | None: 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]: diff --git a/app/pipeline/analyze.py b/app/pipeline/analyze.py index 2acbb3f1..03a81820 100644 --- a/app/pipeline/analyze.py +++ b/app/pipeline/analyze.py @@ -5,7 +5,8 @@ from pathlib import Path from app.core.config import JOBS_DIR, TIMEOUT_ANALYZE, ffmpeg_executable -from app.core.models import Job, _set +from app.core.models import Job, JobCancelled, _set +from app.pipeline.cancel import check_cancel, run_registered logger = logging.getLogger("stemdeck.analyze") @@ -258,13 +259,18 @@ def _load_audio_ffmpeg( sr: int = 22050, duration: float | None = 180.0, timeout: int = TIMEOUT_ANALYZE, + job: Job | None = None, ) -> tuple[object, int] | None: """Decode `source` to a mono float32 numpy array at `sr` via ffmpeg. Bypasses librosa's deprecated audioread fallback (which fires a FutureWarning on .webm/.m4a/.opus inputs because soundfile can't read those directly). `duration=None` decodes the whole file (used by the beat-grid stage, which must cover the full track). Returns - (samples, sr) or None on failure.""" + (samples, sr) or None on failure. + + With `job`, ffmpeg is registered against it so a cancel can terminate the + decode rather than wait it out (#748). A terminated decode reads as a + failure here; the caller's own cancel check is what turns it into one.""" import numpy as np # Defence in depth: even though `source` is constructed by the server @@ -302,7 +308,10 @@ def _load_audio_ffmpeg( cmd += ["-t", str(duration)] # cap input duration cmd.append("-") # write to stdout try: - proc = subprocess.run(cmd, capture_output=True, check=True, timeout=timeout) + if job is not None: + result = run_registered(job, cmd, timeout, capture_stdout=True) + else: + result = subprocess.run(cmd, capture_output=True, timeout=timeout) # OSError, not just FileNotFoundError. A missing binary is only one way # spawning fails: a portable install whose ffmpeg lost its executable bit # raises PermissionError, which is an OSError but not a FileNotFoundError, @@ -312,7 +321,10 @@ def _load_audio_ffmpeg( except (subprocess.SubprocessError, OSError) as e: logger.warning("ffmpeg decode failed for %s: %s", source, e) return None - y = np.frombuffer(proc.stdout, dtype=np.float32) + if result.returncode != 0: + logger.warning("ffmpeg decode failed for %s: exit %s", source, result.returncode) + return None + y = np.frombuffer(result.stdout, dtype=np.float32) if y.size == 0: return None return y, sr @@ -320,7 +332,11 @@ def _load_audio_ffmpeg( def analyze(job: Job, source: Path) -> tuple[int | None, str | None]: """Best-effort BPM and key detection. On failure, returns (None, None) - and leaves job fields untouched -- the chips stay as placeholders.""" + and leaves job fields untouched -- the chips stay as placeholders. + + A cancel is the one thing that is not swallowed: it raises JobCancelled. + The librosa steps cannot be interrupted, so the flag is checked between + them, and the decode is registered so cancel can stop it outright (#748).""" logger.info("analyze: entering for job %s, source=%s", job.id, source) _set(job, status="analyzing", progress=0.0, stage="Analyzing audio...") try: @@ -333,7 +349,8 @@ def analyze(job: Job, source: Path) -> tuple[int | None, str | None]: # Analyse the first 180 s. Decode via ffmpeg directly into numpy # to avoid librosa's deprecated audioread fallback for # .webm/.m4a/.opus inputs. - loaded = _load_audio_ffmpeg(source, sr=22050, duration=180.0) + loaded = _load_audio_ffmpeg(source, sr=22050, duration=180.0, job=job) + check_cancel(job) if loaded is None: return None, None y, sr = loaded @@ -343,8 +360,10 @@ def analyze(job: Job, source: Path) -> tuple[int | None, str | None]: # cleaner pitch profile on the harmonic component (no cymbal # smear, no kick fundamentals leaking in). y_harmonic, y_percussive = librosa.effects.hpss(y) + check_cancel(job) tempo_arr, beat_frames = librosa.beat.beat_track(y=y_percussive, sr=sr) + check_cancel(job) try: tempo = float(tempo_arr[0]) # type: ignore[index] except (TypeError, IndexError): @@ -359,6 +378,7 @@ def analyze(job: Job, source: Path) -> tuple[int | None, str | None]: whole_song = y.size < 179.5 * sr detected = detect_key_from_audio(y_harmonic, None, sr, whole_song=whole_song) key, scale, key_confidence = detected if detected else (None, None, None) + check_cancel(job) # LUFS / peak. Computed on the same 22 kHz mono buffer; this # loses a few dB of accuracy vs full-sample-rate stereo, but @@ -396,6 +416,8 @@ def analyze(job: Job, source: Path) -> tuple[int | None, str | None]: stage="Analysis complete", ) return bpm, key + except JobCancelled: + raise except Exception: # Full traceback goes to the log; the UI stage line stays generic -- # raw exception reprs (paths, library internals) must not reach it. @@ -490,7 +512,7 @@ def refine_key_from_stems(job: Job, stems_dir: Path) -> None: path = stems_dir / f"{name}.wav" if not path.is_file(): continue - loaded = _load_audio_ffmpeg(path, sr=sr, duration=_STEM_KEY_MAX_SECONDS) + loaded = _load_audio_ffmpeg(path, sr=sr, duration=_STEM_KEY_MAX_SECONDS, job=job) if loaded is None: continue y = loaded[0] * weight @@ -503,7 +525,7 @@ def refine_key_from_stems(job: Job, stems_dir: Path) -> None: bass = None bass_path = stems_dir / "bass.wav" if bass_path.is_file(): - loaded = _load_audio_ffmpeg(bass_path, sr=sr, duration=_STEM_KEY_MAX_SECONDS) + loaded = _load_audio_ffmpeg(bass_path, sr=sr, duration=_STEM_KEY_MAX_SECONDS, job=job) if loaded is not None: bass = loaded[0] harmony, bass = _pad_pair(harmony, bass) diff --git a/app/pipeline/cancel.py b/app/pipeline/cancel.py new file mode 100644 index 00000000..984ab899 --- /dev/null +++ b/app/pipeline/cancel.py @@ -0,0 +1,53 @@ +"""What a pipeline stage needs so a cancel can actually stop it. + +Lives apart from runner.py so analyze.py can use it too: runner imports +analyze, so analyze reaching back into runner would be a circular import. The +analyze stage had no cancel path at all, and on a long track it runs for over +a minute (85 s in one report), so a song trashed or cancelled there kept +"Analyzing" until it finished (#748). +""" + +from __future__ import annotations + +import subprocess + +from app.core.models import Job, JobCancelled +from app.core.registry import set_proc + + +def check_cancel(job: Job) -> None: + if job.cancel_requested: + raise JobCancelled() + + +def run_registered( + job: Job, cmd: list[str], timeout: int, *, capture_stdout: bool = False +) -> subprocess.CompletedProcess[bytes]: + """Run a subprocess with it registered, so cancel can reach it. + + subprocess.run() cannot be interrupted: POST /cancel sets the flag, but + nothing looks at it until the call returns, so a cancel during a large + upload's transcode was a no-op for up to TIMEOUT_FFMPEG per call -- twice + over on the .mp4 path, which runs both this and the video extract (#519). + The analyze decode had the same hole (#748). + + Mirrors collect._run_ffmpeg, which registers for exactly this reason. A + terminated process comes back as a non-zero returncode, never an + exception; a timeout kills it and raises subprocess.TimeoutExpired. + """ + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE if capture_stdout else subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + set_proc(job.id, proc) + try: + try: + stdout, stderr = proc.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + raise + return subprocess.CompletedProcess(cmd, proc.returncode, stdout or b"", stderr or b"") + finally: + set_proc(job.id, None) diff --git a/app/pipeline/collect.py b/app/pipeline/collect.py index 0b6fd1f1..f995ac3c 100644 --- a/app/pipeline/collect.py +++ b/app/pipeline/collect.py @@ -15,7 +15,7 @@ TIMEOUT_FFMPEG, ffmpeg_executable, ) -from app.core.models import Job +from app.core.models import FINISHED_STATUSES, Job from app.core.registry import all_jobs as registry_all from app.core.registry import persist as registry_persist from app.core.registry import remove as registry_remove @@ -68,7 +68,29 @@ def _run_ffmpeg(job: Job, cmd: list[str]) -> bool: set_proc(job.id, None) -_TERMINAL = frozenset(("done", "error", "cancelled")) +def remove_job_dir(job_dir: Path) -> bool: + """Remove a job's directory. False means files are still on disk. + + 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. On + Windows it is a handle a process that just exited has not quite let go of, + which is what the runner's cancel path meets right after terminating + ffmpeg or the demucs worker (#749). Shared by delete_job and that path.""" + 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() + + +_TERMINAL = FINISHED_STATUSES def collect(job: Job, stems_root: Path, job_dir: Path) -> list[str]: diff --git a/app/pipeline/jobqueue.py b/app/pipeline/jobqueue.py index ae872b23..9e61dd91 100644 --- a/app/pipeline/jobqueue.py +++ b/app/pipeline/jobqueue.py @@ -26,7 +26,7 @@ from pathlib import Path from app.core.config import JOBS_DIR -from app.core.models import Job, _set +from app.core.models import FINISHED_STATUSES, Job, _set, settle_cancelled from app.core.registry import get as registry_get from app.core.registry import persist as registry_persist @@ -208,10 +208,10 @@ def _finalise_dropped_job(job: Job) -> None: running slot, so a cancel arriving there sets cancel_requested and returns. The worker owns the job by then and is the only thing that can close it out (#520).""" - if not job.cancel_requested or job.status in ("done", "error", "cancelled"): + if not job.cancel_requested or job.status in FINISHED_STATUSES: return - _set(job, status="cancelled", stage="Cancelled") - cleanup_job_dir(job.id) + if settle_cancelled(job): + cleanup_job_dir(job.id) registry_persist(JOBS_DIR) @@ -234,7 +234,7 @@ async def _worker_loop() -> None: job = registry_get(job_id) if job is None: continue - if job.cancel_requested or job.status in ("done", "error", "cancelled"): + if job.cancel_requested or job.status in FINISHED_STATUSES: # Cancelled or finished while it waited. # # Finalising here is not optional. Between _pop_next() above and diff --git a/app/pipeline/runner.py b/app/pipeline/runner.py index ab484902..4f06fcec 100644 --- a/app/pipeline/runner.py +++ b/app/pipeline/runner.py @@ -17,18 +17,21 @@ LYRICS_LOOKUP_GRACE_SEC, TIMEOUT_FFMPEG, ) -from app.core.models import Job, JobCancelled, _set +from app.core.models import Job, JobCancelled, _set, settle_cancelled from app.core.redact import redact -from app.core.registry import is_upload, set_proc +from app.core.registry import is_upload from app.core.registry import persist as persist_registry from app.pipeline.analyze import analyze, refine_key_from_stems from app.pipeline.beatgrid import compute_beat_grid +from app.pipeline.cancel import check_cancel as _check_cancel +from app.pipeline.cancel import run_registered from app.pipeline.collect import ( cleanup_source, collect, compute_stem_peaks, make_original_track, make_selected_mix, + remove_job_dir, ) from app.pipeline.download import download from app.pipeline.errors import classify_failure @@ -55,35 +58,6 @@ def _rmtree(path: Path) -> None: _pipeline_lock = asyncio.Semaphore(1) -def _check_cancel(job: Job) -> None: - if job.cancel_requested: - raise JobCancelled() - - -def _run_registered_ffmpeg(job: Job, cmd: list[str], timeout: int) -> tuple[int, bytes]: - """Run ffmpeg with the process registered, so cancel can reach it. - - subprocess.run() cannot be interrupted: POST /cancel sets the flag, but - nothing looks at it until the call returns, so a cancel during a large - upload's transcode was a no-op for up to TIMEOUT_FFMPEG per call -- twice - over on the .mp4 path, which runs both this and the video extract (#519). - - Mirrors collect._run_ffmpeg, which registers for exactly this reason. - """ - proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) - set_proc(job.id, proc) - try: - try: - _, stderr = proc.communicate(timeout=timeout) - except subprocess.TimeoutExpired: - proc.kill() - proc.communicate() - raise - return proc.returncode, stderr or b"" - finally: - set_proc(job.id, None) - - def _extract_video_track(job: Job, source: Path, job_dir: Path) -> None: """For an .mp4 upload, preserve a silent video-only track at video.mp4 so the studio can later mux it with a custom stem mix @@ -111,7 +85,7 @@ def _extract_video_track(job: Job, source: Path, job_dir: Path) -> None: str(dest), ] try: - returncode, _ = _run_registered_ffmpeg(job, cmd, TIMEOUT_FFMPEG) + returncode = run_registered(job, cmd, TIMEOUT_FFMPEG).returncode except (OSError, subprocess.SubprocessError) as e: # ffmpeg missing or timed out. Distinct from an .mp4 that simply has no # video stream, and the only one of the two worth surfacing (#436). @@ -162,10 +136,10 @@ def _prepare_local_source(job: Job, source: Path, job_dir: Path) -> Path: "-y", str(dest), ] - returncode, stderr = _run_registered_ffmpeg(job, cmd, TIMEOUT_FFMPEG) - if returncode != 0: + result = run_registered(job, cmd, TIMEOUT_FFMPEG) + if result.returncode != 0: raise RuntimeError( - "ffmpeg transcode failed: " + stderr.decode("utf-8", errors="replace").strip() + "ffmpeg transcode failed: " + result.stderr.decode("utf-8", errors="replace").strip() ) source.unlink(missing_ok=True) return dest @@ -495,11 +469,23 @@ async def _run_async( " (wrapped)" if not isinstance(e, JobCancelled) else "", job.id, ) - _set(job, status="cancelled", stage="Cancelled") + remove_files = settle_cancelled(job) persist_registry(jobs_dir) release_source(job.id) - _rmtree(job_dir) + # A job stopped from the Trash keeps its files until the Trash is + # emptied (#748). A plain cancel's folder is retried like delete_job's + # removal: right after the cancel terminated ffmpeg or the demucs + # worker, Windows can still hold one of its files, and a single attempt + # left the folder behind for good. + if remove_files and not remove_job_dir(job_dir): + logger.warning("[%s] cancelled job's folder could not be removed", job.id) return + # A cancel or a trash that arrived after the last stage checked for it lost + # the race, and the job finished anyway. Left set, the flags would make + # every later vocal split, section run or lyric timing on this done job + # cancel itself at once, until the next restart (#748). + job.cancel_requested = False + job.stop_requested = False _set(job, status="done", progress=1.0, stage="Done") _write_metadata(job, job_dir) persist_registry(jobs_dir) diff --git a/desktop/src-tauri/src/main.rs b/desktop/src-tauri/src/main.rs index 442e0c0f..735bdaa2 100644 --- a/desktop/src-tauri/src/main.rs +++ b/desktop/src-tauri/src/main.rs @@ -361,6 +361,32 @@ fn main() { .plugin(tauri_plugin_dialog::init()) .plugin(tauri_plugin_store::Builder::default().build()) .setup(|app| { + // A portable folder the user cannot write to (Program Files, say) + // cannot work at all: setup, the CUDA swap and the updater all + // write inside it. Say so and stop, before the window opens and + // before anything touches the install (#746). + #[cfg(any(windows, target_os = "linux"))] + if let Ok(root) = app_root() { + let outcome = + portable_root_check(is_portable_package(&root), || app_root_is_writable(&root)); + if outcome == PortableRootCheck::NotWritable { + use tauri_plugin_dialog::DialogExt; + eprintln!( + "[stemdeck] portable folder is not writable: {}", + root.display() + ); + // Non-blocking show: the setup hook runs on the main + // thread, and the callback ends the process once the + // user has read the message. + app.dialog() + .message(not_writable_message(&root)) + .title("StemDeck cannot start here") + .kind(tauri_plugin_dialog::MessageDialogKind::Error) + .show(|_| std::process::exit(0)); + return Ok(()); + } + } + // The main window is built here rather than by Tauri from // tauri.conf.json, because one of its settings has to differ by // platform and the config file has no way to say so. The config @@ -1254,6 +1280,45 @@ fn app_root_is_writable(root: &Path) -> bool { } } +/// What the startup check decided about the app root (#746). +#[cfg(any(windows, target_os = "linux"))] +#[derive(Debug, PartialEq, Eq)] +enum PortableRootCheck { + Proceed, + NotWritable, +} + +/// A portable package keeps its runtime, torch swap and updates inside its own +/// folder, so it is only usable where the user can write. Installs that are not +/// portable keep their data elsewhere and are never stopped by this. +/// +/// The probe is a closure so it only runs for a portable package. It creates a +/// file in the app root, and on any other install, a root-owned /opt one in +/// particular, writing there at every start is neither needed nor safe. +#[cfg(any(windows, target_os = "linux"))] +fn portable_root_check(portable: bool, probe_writable: impl FnOnce() -> bool) -> PortableRootCheck { + if portable && !probe_writable() { + PortableRootCheck::NotWritable + } else { + PortableRootCheck::Proceed + } +} + +/// Plain wording, with the folder shown. Elevation is deliberately not offered: +/// an elevated window cannot receive drag and drop from a normal Explorer +/// (Windows UIPI), and every launch would prompt. +#[cfg(any(windows, target_os = "linux"))] +fn not_writable_message(root: &Path) -> String { + format!( + "StemDeck cannot write to the folder it is in:\n\n{}\n\n\ + A portable StemDeck must live in a folder you can write to, for example \ + C:\\StemDeck. It cannot run from Program Files, and Documents or Desktop \ + can refuse it too when Windows security protects those folders.\n\n\ + Move the whole folder there and start StemDeck again.", + root.display() + ) +} + /// Stops the backend and waits for the process to actually exit. /// /// The regular `stop_backend` hands the kill to a background thread and @@ -6948,6 +7013,35 @@ mod tests { assert_eq!(fs::read_dir(dir.path()).unwrap().count(), 0); } + // #746: a portable zip extracted under Program Files is not writable + // without admin. Only that combination stops startup. + #[cfg(any(windows, target_os = "linux"))] + #[test] + fn only_an_unwritable_portable_root_stops_startup() { + use super::{portable_root_check, PortableRootCheck::*}; + assert_eq!(portable_root_check(true, || false), NotWritable); + assert_eq!(portable_root_check(true, || true), Proceed); + assert_eq!(portable_root_check(false, || false), Proceed); + assert_eq!(portable_root_check(false, || true), Proceed); + } + + // The probe writes a file into the app root, so an install that is not + // portable must never run it (security review of #746). + #[cfg(any(windows, target_os = "linux"))] + #[test] + fn a_non_portable_install_is_never_probed() { + let outcome = super::portable_root_check(false, || panic!("probed a non-portable root")); + assert_eq!(outcome, super::PortableRootCheck::Proceed); + } + + #[cfg(any(windows, target_os = "linux"))] + #[test] + fn the_not_writable_message_names_the_folder_and_a_fix() { + let msg = super::not_writable_message(Path::new(r"C:\Program Files (x86)\Stemdeck")); + assert!(msg.contains("Program Files (x86)")); + assert!(msg.contains(r"C:\StemDeck")); + } + // install.sh --global puts the package in /opt, root-owned, while the app // runs as the user. The updater has to decline up front rather than fail // part way through the swap. diff --git a/packaging/windows/README-WINDOWS.txt b/packaging/windows/README-WINDOWS.txt index b17a30e9..9b1ee89a 100644 --- a/packaging/windows/README-WINDOWS.txt +++ b/packaging/windows/README-WINDOWS.txt @@ -4,7 +4,8 @@ StemDeck Windows Portable Alpha Run --- -1. Extract the zip folder. +1. Extract the zip folder to a folder you can write to (for example C:\StemDeck), + not Program Files. 2. Double-click StemDeck.exe. 3. Let first-run setup prepare local runtime assets. diff --git a/static/css/daw.css b/static/css/daw.css index b4527f1f..db6cde71 100644 --- a/static/css/daw.css +++ b/static/css/daw.css @@ -929,6 +929,16 @@ input, textarea { font-family: inherit; } .cat-del:hover { background: rgba(21,31,39,0.8); color: var(--danger); } .cat-fav:hover { background: rgba(21,31,39,0.8); color: var(--fg); } .cat-fav.active, .cat-fav.active:hover { color: #e54e4e; } +/* A stopped song (#748) offers Extract at rest: it is the row's only way on. */ +.cat-item.is-stopped .cat-actions { opacity: 1; pointer-events: auto; } +.cat-extract { + height: 20px; padding: 0 8px; border: 1px solid var(--accent); border-radius: 5px; + background: transparent; color: var(--accent); cursor: pointer; + font: inherit; font-size: 11px; white-space: nowrap; flex-shrink: 0; + transition: background var(--t-fast), color var(--t-fast); +} +.cat-extract:hover, .cat-extract:focus-visible { background: var(--accent); color: var(--panel-2); } +.cat-extract:disabled { opacity: 0.5; cursor: default; } /* A favourite says so at rest, in the subline, where it takes no title room. */ .cat-item.is-fav .cat-sub::before { content: ""; flex-shrink: 0; align-self: center; diff --git a/static/js/catalog.js b/static/js/catalog.js index 8fae4f6a..51e68275 100644 --- a/static/js/catalog.js +++ b/static/js/catalog.js @@ -2,14 +2,14 @@ import { STEM_NAMES } from "./constants.js"; import { wireUpAudio, updateFooterTrack } from "./player.js"; import { initSections } from "./sections.js"; -import { bpmChip, foregroundJobId, keyChip, saveSelectedStems, selectedStems, titleEl } from "./state.js"; +import { audioEngine, bpmChip, foregroundJobId, keyChip, multitrack, saveSelectedStems, selectedStems, titleEl } from "./state.js"; import { refreshStemChoiceVisuals } from "./stemChoice.js"; import { trackFormat, formatIconSvg, paintNowPlayingArt } from "./formatIcon.js"; import { formatKey } from "./keyLabel.js"; import { showError, importFromUrl, detachForegroundJob, runVocalSplitIfWanted } from "./job.js"; import { - cancelQueuedJob, getQueueSnapshot, isPaused, onJobSettled, onQueueChange, - queueCount, queueRowStates, reorderQueuedJob, runningLabel, startQueue, + cancelQueuedJob, currentIds, getQueueSnapshot, isPaused, onJobSettled, onQueueChange, + queueCount, queueRowStates, refreshQueue, reorderQueuedJob, runningLabel, startQueue, startQueueStream, } from "./queue.js"; import { fmtTime, isReimportableSource, storeGet, storeSet } from "./utils.js"; @@ -211,6 +211,9 @@ let _currentTrackId = null; // The open track's tags as the last render() saw them. See setCurrentTrack(). let _renderedTagsKey = ""; let _loadTrackToken = 0; +// Imports submitted from the form, mapped to the load token at that moment. A +// finished one opens in the studio only if nothing has been opened since (#747). +const _autoOpenTokens = new Map(); let catalogView = "library"; let catalogSearchQuery = ""; @@ -220,6 +223,13 @@ const TRASH_ID = "trash"; // The default landing folder for unorganized tracks — protected from deletion. const UNSORTED_ID = "f-unsorted"; const PROCESSING_STATUSES = new Set(["queued", "downloading", "analyzing", "separating", "processing"]); + +// One trash can for every "throw this away" button: the library rows and the +// queue rows (#748), so the queue's cancel reads as the same action. +const TRASH_ICON_SVG = ``; const FOLDER_COLORS = ["#d8a84a", "#e85f6f", "#64c86f", "#4f9de8", "#a985f4"]; const DEFAULT_FOLDER_COLOR = FOLDER_COLORS[0]; const TRACK_DRAG_TYPE = "application/x-stemdeck-track"; @@ -690,6 +700,8 @@ export function updateTrackStatus(trackId, status) { for (const el of document.querySelectorAll(`.cat-item[data-id="${trackId}"]`)) { el.classList.toggle("unavailable", status === "unavailable"); } + // A stopped row changes its line and gains Extract (#748), more than a dot. + if (status === "stopped") render(); } } @@ -938,7 +950,7 @@ function applyTrackInfoToPanel(track) { */ function syncTrashToServer(trackId, trashed) { const action = trashed ? "trash" : "restore"; - fetch(`/api/jobs/${encodeURIComponent(trackId)}/${action}`, { method: "POST" }) + return fetch(`/api/jobs/${encodeURIComponent(trackId)}/${action}`, { method: "POST" }) .catch((e) => console.warn(`[catalog] could not ${action} ${trackId} on the server`, e)); } @@ -987,7 +999,13 @@ function moveTrackToTrash(trackId) { // purgeTrash() does the same on permanent delete, for a job that errors // after being trashed but before it's purged. dismissFailuresByJobId(trackId); - syncTrashToServer(trackId, true); + // The server stops a job that is still importing when it is trashed, its + // files kept (#748). Ask for the queue again once it has, so the row goes now rather + // than whenever the next frame happens to arrive. + const wasInQueue = currentIds(getQueueSnapshot()).has(trackId); + syncTrashToServer(trackId, true).then(() => { + if (wasInQueue) refreshQueue(); + }); saveState(); render(); } @@ -1684,6 +1702,8 @@ function trackSublineHtml(track, { inTrash = false } = {}) { parts = [i18nT("job.processing")]; } else if (track.status === "error") { parts = [i18nT("notifKind.importFailed")]; + } else if (track.status === "stopped") { + parts = [i18nT("track.stopped")]; } else { // Just the length. The stem count was here too, but every row carrying // "6 stems" says the same thing forty times over, and the now-playing card @@ -1768,13 +1788,43 @@ function makeStripItem({ className = "", id, title, html, color, trackId }) { return item; } +// A song stopped by the Trash keeps its files (#748). Restore brings it back +// as it was, and this is the one way to run it again: the user decides when it +// gets the GPU, never Restore and never a reload. +function extractButtonHtml(track) { + const label = i18nT("track.extractAgainOf", { title: displayTitle(track.title) }); + return ``; +} + +async function extractStoppedTrack(trackId) { + try { + const res = await fetch(`/api/jobs/${encodeURIComponent(trackId)}/extract`, { method: "POST" }); + if (!res.ok) { + const body = await res.json().catch(() => ({})); + notifyFailure({ kind: "import", message: body.detail || i18nT("job.audioProcessingFailed") }); + render(); + return; + } + if (tracks[trackId]) tracks[trackId].status = "queued"; + // Like any import (#747): it opens when done, if the studio is still empty. + armAutoOpen(trackId); + saveState(); + render(); + await refreshQueue(); + } catch (e) { + console.warn("[catalog] could not extract", trackId, e); + render(); + } +} + function renderTrackItem(trackId, { inTrash = false } = {}) { const track = tracks[trackId]; if (!track) return null; const el = document.createElement("div"); const isUnavailable = track.status === "unavailable"; - el.className = `cat-item${trackId === _currentTrackId ? " active" : ""}${isUnavailable ? " unavailable" : ""}${!inTrash && track.favorite ? " is-fav" : ""}`; + el.className = `cat-item${trackId === _currentTrackId ? " active" : ""}${isUnavailable ? " unavailable" : ""}${!inTrash && track.favorite ? " is-fav" : ""}${!inTrash && track.status === "stopped" ? " is-stopped" : ""}`; el.dataset.id = trackId; el.innerHTML = ` @@ -1784,11 +1834,8 @@ function renderTrackItem(trackId, { inTrash = false } = {}) {
${trackSublineHtml(track, { inTrash })}
- ${inTrash ? "" : `
${favButtonHtml(track)}
`} `; el.querySelector(".cat-del")?.setAttribute("aria-label", i18nT("track.moveTitleToTrash", { title: track.title ?? i18nT("track.unknown") })); @@ -1797,6 +1844,11 @@ function renderTrackItem(trackId, { inTrash = false } = {}) { e.stopPropagation(); moveTrackToTrash(trackId); }); + el.querySelector(".cat-extract")?.addEventListener("click", (e) => { + e.stopPropagation(); + e.currentTarget.disabled = true; + extractStoppedTrack(trackId); + }); wireFavButton(el, trackId); wireTrackDragAndLoad(el, trackId); @@ -2232,6 +2284,25 @@ function decorateRow(el, rowState) { if (fill) fill.style.width = `${Math.round(rowState.progress * 100)}%`; } +/** Remember that this import was started with the studio as it is now. Imports + * never take the studio while they run (#747); this is what lets the first one + * to finish open itself, but only while the studio is still empty and the user + * has not opened anything since. */ +export function armAutoOpen(jobId) { + _autoOpenTokens.set(jobId, _loadTrackToken); +} + +/** True when the studio shows no track and no load is under way. Either engine + * counts: the Web Audio path sets audioEngine, the streaming path sets + * multitrack. A load in flight has bumped the token before it has an engine. + * "Sync again" holds the studio through foregroundJobId without either: it + * tears the player down and shows its own overlay, so it counts as busy too. */ +function shouldAutoOpen(jobId) { + const armedAt = _autoOpenTokens.get(jobId); + _autoOpenTokens.delete(jobId); + return armedAt === _loadTrackToken && !audioEngine && !multitrack && !foregroundJobId; +} + /** A background import has finished (or failed, or was cancelled). It has no * per-job stream, so fetch its final state once and complete its library entry * -- stems, duration and analysis all land here, which is what makes the track @@ -2248,6 +2319,7 @@ async function completeSettledJob(jobId) { } const state = await res.json(); if (state.status === "cancelled") { + _autoOpenTokens.delete(jobId); // Nothing was produced; drop the placeholder row rather than leaving a // track that can never be loaded. delete tracks[jobId]; @@ -2282,7 +2354,14 @@ async function completeSettledJob(jobId) { const finalState = state.status === "done" ? await runVocalSplitIfWanted(state) : state; const track = stateMetadataToTrack(finalState, { ...existing, id: jobId }); track.id = jobId; + // A server that sends "work" looked for it while the job ran, so the track + // that opens next never needs to ask again (#747: every import lands here). + if (finalState.status === "done" && "work" in finalState) track.workChecked = true; addTrackToLibrary(track); + // Checked after the split so the stems it makes are there when it opens, + // and through the library's own loader so it behaves like a click on the row. + if (finalState.status === "done" && shouldAutoOpen(jobId)) loadTrackIntoStudio(jobId); + else _autoOpenTokens.delete(jobId); } catch (e) { console.warn("[catalog] could not finish background job", jobId, e); } @@ -2329,9 +2408,7 @@ function queueRowHtml({ job, running }, place, { paused = false } = {}) { `} `; } @@ -3540,9 +3617,26 @@ async function syncWithServer() { if (favBtn && current) paintFavButton(favBtn, Boolean(current.favorite)); } reconcileAvailability(jobs); + settleStaleProcessingTracks(); } catch (e) { console.warn("[catalog] failed to load jobs from backend:", e); } } +/** + * Ask the server where every track this side still calls "processing" stands. + * + * A job is only settled here when it leaves the queue while the page watches + * it. One that stopped, failed or finished while the app was closed never left + * the queue in front of this page, so its row said "processing" for good: the + * song trashed mid-extraction in #748 sat in the Trash that way after a + * restart. completeSettledJob takes whatever the server says, so a job that + * really is still waiting just keeps its status. + */ +function settleStaleProcessingTracks() { + for (const [id, track] of Object.entries(tracks)) { + if (PROCESSING_STATUSES.has(track.status)) completeSettledJob(id); + } +} + // ─── Settings menu + Library editor ─── let libraryEditor = null; @@ -4903,7 +4997,7 @@ async function waitForJobTerminal(jobId) { const r = await fetch(`/api/jobs/${jobId}`, { cache: "no-store" }); if (r.status === 404) return; const s = await r.json(); - if (s.status === "done" || s.status === "error" || s.status === "cancelled") return; + if (["done", "error", "cancelled", "stopped"].includes(s.status)) return; } catch { /* transient — keep waiting */ } } } diff --git a/static/js/i18n.js b/static/js/i18n.js index 174f2289..c746b052 100644 --- a/static/js/i18n.js +++ b/static/js/i18n.js @@ -805,6 +805,9 @@ const en = { "track.untitled": "Untitled track", "track.unknown": "Unknown track", "track.removed": "Removed", + "track.stopped": "Stopped", + "track.extractAgain": "Extract", + "track.extractAgainOf": "Extract {title} again", "track.moveToTrash": "Move to Trash", "track.moveTitleToTrash": "Move {title} to Trash", "track.favorite": "Add to Favorites", @@ -1614,6 +1617,9 @@ const pl = { "track.untitled": "Utwór bez tytułu", "track.unknown": "Nieznany utwór", "track.removed": "Usunięto", + "track.stopped": "Zatrzymano", + "track.extractAgain": "Wyodrębnij", + "track.extractAgainOf": "Wyodrębnij ponownie {title}", "track.moveToTrash": "Przenieś do kosza", "track.moveTitleToTrash": "Przenieś {title} do kosza", "track.favorite": "Dodaj do ulubionych", @@ -2402,6 +2408,9 @@ const ja = { "track.untitled": "無題のトラック", "track.unknown": "不明なトラック", "track.removed": "削除済み", + "track.stopped": "停止済み", + "track.extractAgain": "抽出", + "track.extractAgainOf": "{title} を再抽出", "track.moveToTrash": "ゴミ箱に移動", "track.moveTitleToTrash": "{title}をゴミ箱に移動", "track.favorite": "お気に入りに追加", @@ -3174,6 +3183,9 @@ const zhHans = { "track.untitled": "未命名曲目", "track.unknown": "未知曲目", "track.removed": "已移除", + "track.stopped": "已停止", + "track.extractAgain": "提取", + "track.extractAgainOf": "重新提取 {title}", "track.moveToTrash": "移到回收站", "track.moveTitleToTrash": "将 {title} 移到回收站", "track.favorite": "添加到收藏", @@ -3951,6 +3963,9 @@ const de = { "track.untitled": "Unbenannter Track", "track.unknown": "Unbekannter Track", "track.removed": "Entfernt", + "track.stopped": "Gestoppt", + "track.extractAgain": "Extrahieren", + "track.extractAgainOf": "{title} erneut extrahieren", "track.moveToTrash": "In den Papierkorb verschieben", "track.moveTitleToTrash": "{title} in den Papierkorb verschieben", "track.favorite": "Zu Favoriten hinzufügen", @@ -4734,6 +4749,9 @@ const pt = { "track.untitled": "Faixa sem título", "track.unknown": "Faixa desconhecida", "track.removed": "Removido", + "track.stopped": "Interrompida", + "track.extractAgain": "Extrair", + "track.extractAgainOf": "Extrair {title} novamente", "track.moveToTrash": "Mover para a lixeira", "track.moveTitleToTrash": "Mover {title} para a lixeira", "track.favorite": "Adicionar aos favoritos", @@ -5514,6 +5532,9 @@ const id = { "track.untitled": "Trek tanpa judul", "track.unknown": "Trek tidak dikenal", "track.removed": "Dihapus", + "track.stopped": "Dihentikan", + "track.extractAgain": "Ekstrak", + "track.extractAgainOf": "Ekstrak ulang {title}", "track.moveToTrash": "Pindahkan ke Sampah", "track.moveTitleToTrash": "Pindahkan {title} ke Sampah", "track.favorite": "Tambahkan ke favorit", @@ -6291,6 +6312,9 @@ const fr = { "track.untitled": "Morceau sans titre", "track.unknown": "Morceau inconnu", "track.removed": "Supprimé", + "track.stopped": "Arrêté", + "track.extractAgain": "Extraire", + "track.extractAgainOf": "Extraire à nouveau {title}", "track.moveToTrash": "Mettre à la corbeille", "track.moveTitleToTrash": "Mettre {title} à la corbeille", "track.favorite": "Ajouter aux favoris", @@ -7249,6 +7273,9 @@ const es = { "track.untitled": "Pista sin título", "track.unknown": "Pista desconocida", "track.removed": "Eliminada", + "track.stopped": "Detenida", + "track.extractAgain": "Extraer", + "track.extractAgainOf": "Volver a extraer {title}", "track.moveToTrash": "Mover a la papelera", "track.moveTitleToTrash": "Mover {title} a la papelera", "track.favorite": "Añadir a favoritos", @@ -8049,6 +8076,9 @@ const ko = { "track.untitled": "제목 없는 트랙", "track.unknown": "알 수 없는 트랙", "track.removed": "삭제됨", + "track.stopped": "중지됨", + "track.extractAgain": "추출", + "track.extractAgainOf": "{title} 다시 추출", "track.moveToTrash": "휴지통으로 옮기기", "track.moveTitleToTrash": "{title}을(를) 휴지통으로 옮기기", "track.favorite": "즐겨찾기에 추가", diff --git a/static/js/job.js b/static/js/job.js index e6bd9e53..a74e615f 100644 --- a/static/js/job.js +++ b/static/js/job.js @@ -3,13 +3,12 @@ import { jobDetailEl, jobCancelBtn, progressEl, titleEl, bpmChip, keyChip, eventSource, setEventSource, setCurrentJobId, foregroundJobId, setForegroundJobId, - audioEngine, multitrack, selectedStems, vocalSplitMode, } from "./state.js"; import { destroyPlayer, wireUpAudio, setWaveformLoading, updateFooterTrack } from "./player.js"; import { notifyFailure, dismissFailuresByJobId } from "./notifications.js"; import { getStagePhrases } from "./phrases.js"; -import { addTrackToLibrary, setCurrentTrack, updateTrackStatus, applyStemPresenceCards, libraryAudioTags, libraryArtist, libraryIdentity, libraryWork, paintFinishedTrackNames } from "./catalog.js"; +import { addTrackToLibrary, setCurrentTrack, updateTrackStatus, applyStemPresenceCards, libraryAudioTags, libraryArtist, libraryIdentity, libraryWork, paintFinishedTrackNames, armAutoOpen } from "./catalog.js"; import { initSections } from "./sections.js"; import { importPlaylist, looksLikePlaylist } from "./playlist.js"; import { t } from "./i18n.js"; @@ -86,7 +85,7 @@ export async function runVocalSplitIfWanted(state) { } } -const TERMINAL_STATUSES = new Set(["done", "error", "cancelled"]); +const TERMINAL_STATUSES = new Set(["done", "error", "cancelled", "stopped"]); // Last library-visible values written per job, so a 4 Hz progress stream does // not re-run addTrackToLibrary (a full localStorage write plus a whole-sidebar @@ -120,13 +119,6 @@ function setSubmitProcessing(processing) { if (label) label.textContent = processing ? t("job.processing") : t("process.extractStems"); } -/** True when audio is loaded in the studio. Either engine counts: the Web Audio - * path sets audioEngine, the streaming path sets multitrack, and destroyPlayer - * clears both. Read at call time so the live bindings are current. */ -function studioHasTrack() { - return !!(audioEngine || multitrack); -} - function pickPhrase(status) { const stagePhrases = getStagePhrases(); const pool = stagePhrases[status] || stagePhrases.default; @@ -506,9 +498,11 @@ function applyState(state) { }, }); setForegroundJobId(null); - } else if (state.status === "cancelled") { + } else if (state.status === "cancelled" || state.status === "stopped") { + // "stopped" is a job put in the Trash while it ran (#748): it ends here the + // same way, but its row keeps saying so, since its files are kept. stopJobPolling(); - updateTrackStatus(state.job_id, "cancelled"); + updateTrackStatus(state.job_id, state.status); setWaveformLoading(false); jobBox.classList.add("hidden"); setForegroundJobId(null); @@ -670,6 +664,7 @@ function registerUploadRow(jobId, file) { peakDb: null, sourceUrl, }); + armAutoOpen(jobId); } function sanitizeFilename(name) { @@ -747,24 +742,15 @@ export function wireJobForm() { form.addEventListener("submit", async (e) => { e.preventDefault(); - // An import must never take a loaded studio away from the user. With a - // track playing, the new job goes straight to the background: no player - // teardown, no loading overlay, no takeover when it finishes. It reports - // progress on its library row instead. + // An import never takes the studio, loaded or empty (#747). It runs in the + // background and reports on its library row and the queue; the card and the + // loading overlay stay as they are. If the studio is still empty when it + // finishes, catalog.js opens it then (see armAutoOpen). // - // An import already holding the foreground keeps it, too. Otherwise - // queueing a second track would point the studio overlay at a job that has - // not started, and the first import -- the one about to finish -- would no - // longer be the one that loads. - const background = studioHasTrack() || !!foregroundJobId; - if (background) { - // Deliberately NOT resetImportUi(): that closes the running import's - // event stream and drops its foreground claim, which would leave the job - // about to finish with nothing listening for its completion. - clearImportError(); - } else { - reset(); - } + // Deliberately NOT resetImportUi(): that closes the event stream of an + // import still holding the foreground (a "Sync again" restore), which would + // leave the job about to finish with nothing listening for its completion. + clearImportError(); setSubmitProcessing(true); const fileInput = document.getElementById("fileInput"); @@ -783,12 +769,10 @@ export function wireJobForm() { } // Several files dropped at once: upload them one after another. Parallel // uploads of several 400 MB bodies would thrash memory on both ends, and - // the endpoint takes exactly one file per request anyway. The first one - // takes the studio if it is free, exactly as a single import would; the - // rest queue behind it. + // the endpoint takes exactly one file per request anyway. All of them run + // in the background; the first to finish opens if the studio is still empty. const batch = fileInput?._files ?? null; if (batch && batch.length > 1) { - if (!background) setWaveformLoading(true, "Uploading…"); let queued = 0; let failure = null; for (const item of batch) { @@ -796,16 +780,6 @@ export function wireJobForm() { const id = await postFileJob(item); registerUploadRow(id, item); queued += 1; - if (queued === 1 && !background) { - setCurrentJobId(id); - setForegroundJobId(id); - setCurrentTrack(id); - jobBox.classList.add("hidden"); - jobCancelBtn.classList.add("hidden"); - startPhraseRotation("queued"); - lastStatus = "queued"; - connectEvents(id); - } } catch (err) { failure = err; break; // a full queue will reject the rest too; stop asking @@ -814,10 +788,6 @@ export function wireJobForm() { setSubmitProcessing(false); fileInput._clear?.(); if (failure) { - if (!queued && !background) { - setWaveformLoading(false); - setForegroundJobId(null); - } showError( `Queued ${queued} of ${batch.length} files: ${failure.message}`, null, @@ -844,20 +814,6 @@ export function wireJobForm() { ? submitBtn.dataset.resplitTitle || t("job.processingTrackTitle") : (sanitized ?? (urlInput.value || t("job.processingTrackTitle"))); - const postUrlText = document.getElementById("post-url-text"); - if (postUrlText) postUrlText.textContent = displayTitle; - - // Show overlay immediately for both paths. File uploads show "Uploading…" - // in the overlay phrase until the fetch completes and SSE takes over. - // Skipped entirely for a background import -- the overlay covers the - // studio, which is exactly what must not happen here. - if (!background) { - setWaveformLoading(true, file ? "Uploading…" : ""); - if (file) { - lastStatus = "queued"; - } - } - let endpoint = "/api/jobs"; let fetchInit; if (resplitJob) { @@ -898,7 +854,6 @@ export function wireJobForm() { jobId = data.job_id; if (data.source_url) newSourceUrl = data.source_url; } catch (err) { - if (file) jobBox.classList.add("hidden"); showError(t("job.startFailed", { message: err.message })); setSubmitProcessing(false); return; @@ -929,26 +884,10 @@ export function wireJobForm() { sourceUrl: newSourceUrl, }); - if (background) { - // No per-job stream: opening one per queued import would burn through - // the browser's ~6 connections per origin and starve stem loading. The - // shared queue stream drives the row, and catalog.js completes the - // library entry when the job leaves the queue. - if (postUrlText) postUrlText.textContent = ""; - return; - } - - setCurrentJobId(jobId); - setForegroundJobId(jobId); - setCurrentTrack(jobId); - - // Keep job box hidden, overlay drives the UI. Start phrase rotation now - // that the job exists on the server. - jobBox.classList.add("hidden"); - jobCancelBtn.classList.add("hidden"); - startPhraseRotation("queued"); - lastStatus = "queued"; - - connectEvents(jobId); + // No per-job stream: opening one per queued import would burn through the + // browser's ~6 connections per origin and starve stem loading. The shared + // queue stream drives the row, and catalog.js completes the library entry + // when the job leaves the queue. + armAutoOpen(jobId); }); } diff --git a/static/mobile/app.js b/static/mobile/app.js index 7bcf4f1e..de4aca92 100644 --- a/static/mobile/app.js +++ b/static/mobile/app.js @@ -591,8 +591,10 @@ function extractProgressCard() { const pct = Math.round((j.progress || 0) * 100); const failed = j.status === "error"; const cancelled = j.status === "cancelled"; + // Trashed while it ran (#748): halted, its files kept for the desktop's Trash. + const stopped = j.status === "stopped"; const done = j.status === "done"; - const line = failed ? "Failed" : cancelled ? "Cancelled" : done ? "Done" : `${j.stage || j.status}…`; + const line = failed ? "Failed" : cancelled ? "Cancelled" : stopped ? "Stopped" : done ? "Done" : `${j.stage || j.status}…`; return `
IN PROGRESS
@@ -718,7 +720,7 @@ function _onExtractState(jobId, s) { state.extractJob.progress = s.progress || 0; state.extractJob.stage = s.stage || s.status; if (state.tab === "extract") render(); - if (s.status === "done" || s.status === "error" || s.status === "cancelled") { + if (["done", "error", "cancelled", "stopped"].includes(s.status)) { if (s.status === "done" && state.extractJob.vocalSplitMode === "split") { _runVocalSplitThenFinish(jobId); } else { diff --git a/tests/e2e/artist-from-server.spec.mjs b/tests/e2e/artist-from-server.spec.mjs index 296a5b93..8c961ef4 100644 --- a/tests/e2e/artist-from-server.spec.mjs +++ b/tests/e2e/artist-from-server.spec.mjs @@ -15,6 +15,7 @@ import { seedCatalogState, stubAudioTags, stubExportEndpoints, + stubImportQueue, stubUpdateCheck, } from "./helpers.mjs"; @@ -106,24 +107,22 @@ test.describe("band found by the server during the import", () => { test("a finished import shows its band at once, with nothing asked of Wikimedia", async ({ page }) => { const asked = await stubWikimedia(page); const doneState = await stubServerBand(page, JOB_ID); - // The import: the server accepts it as the fixture job, and its event - // stream says it is done, carrying the band the pipeline found. + // The import: the server accepts it as the fixture job. It runs in the + // background (#747), so the queue says when it is done and the page then + // fetches its state, which carries the band the pipeline found. await page.route("**/api/jobs", (route) => { if (route.request().method() !== "POST") return route.fallback(); return route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify({ job_id: JOB_ID }) }); }); - await page.route(`**/api/jobs/${JOB_ID}/events`, async (route) => - route.fulfill({ - status: 200, - contentType: "text/event-stream", - body: `data: ${JSON.stringify(await doneState())}\n\n`, - })); - // Only the sibling is in the library, so nothing is open and the import - // takes the studio. + const queue = await stubImportQueue(page, JOB_ID); + // Only the sibling is in the library, so nothing is open and the finished + // import opens itself. await openPage(page, { [SIBLING_JOB_ID]: fixtureTrack(SIBLING_JOB_ID, "E2E Fixture Track (again)") }); await page.locator("#url").fill("https://www.youtube.com/watch?v=dQw4w9WgXcQ"); await page.locator("#submit").click(); + await queue.run(); + await queue.settle(); await expect(page.locator(".app")).not.toHaveClass(/no-track/, { timeout: 15000 }); await expect(page.locator("#np-artist")).toHaveText("Dream Theater", { timeout: 15000 }); diff --git a/tests/e2e/helpers.mjs b/tests/e2e/helpers.mjs index 56eb46f3..f740869b 100644 --- a/tests/e2e/helpers.mjs +++ b/tests/e2e/helpers.mjs @@ -409,3 +409,38 @@ export async function stubLyricsLookup(page, { rows = [], artist = "", song = "" }); return asked; } + +/** + * Stand in for the import queue, for a test that has to watch an import while + * it runs. The e2e backend never separates anything, so a job that stays + * "running" long enough to look at cannot be produced for real. + * + * `run()` reports the job as running and `settle()` as gone, each followed by a + * queue read so the page sees it at once. The queue stream is held off, or its + * own empty frames would override the stub. + */ +export async function stubImportQueue(page, jobId, title = "Importing") { + let running = false; + await page.route("**/api/queue", (route) => + route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ + running: running + ? { job_id: jobId, status: "separating", stage: "Separating 40%", progress: 0.4, title } + : null, + queued: [], + paused: false, + max_pending_uploads: 5, + max_pending_urls: 50, + capacity_left_uploads: 5, + capacity_left_urls: 50, + }), + })); + await page.route("**/api/queue/events", (route) => route.abort()); + const read = () => page.evaluate(async () => (await import("/js/queue.js")).refreshQueue()); + return { + run: async () => { running = true; await read(); }, + settle: async () => { running = false; await read(); }, + }; +} diff --git a/tests/e2e/import-background.spec.mjs b/tests/e2e/import-background.spec.mjs new file mode 100644 index 00000000..a4ec1150 --- /dev/null +++ b/tests/e2e/import-background.spec.mjs @@ -0,0 +1,99 @@ +// An import never takes the studio while it runs (#747). With nothing open it +// used to put the song's name and "Extracting" on the Now Playing card; now the +// card stays empty, the progress lives on the library row and the queue, and the +// finished track opens by itself if the studio is still empty. +// +// The queue is stubbed (see stubImportQueue): the e2e backend never separates. +// The finished job is the seeded fixture, so its stems are real files. +import { test, expect } from "@playwright/test"; + +import { + JOB_ID, + SIBLING_JOB_ID, + TRACK_TITLE, + SIBLING_TITLE, + fixtureTrack, + seedCatalogState, + stubExportEndpoints, + stubImportQueue, + stubUpdateCheck, +} from "./helpers.mjs"; + +test.use({ viewport: { width: 1600, height: 900 } }); + +// Only the sibling is in the store, so the import's own row is new and nothing +// is dedup-folded into it (#542). +async function seedSibling(page) { + await seedCatalogState(page, { + folders: [ + { id: "f-unsorted", name: "Unsorted", items: [SIBLING_JOB_ID], color: null }, + { id: "trash", name: "Trash", items: [], color: null }, + ], + tracks: { [SIBLING_JOB_ID]: fixtureTrack(SIBLING_JOB_ID, SIBLING_TITLE) }, + }); +} + +async function startImport(page) { + await page.route("**/api/jobs", (route) => { + if (route.request().method() !== "POST") return route.fallback(); + return route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify({ job_id: JOB_ID }) }); + }); + const queue = await stubImportQueue(page, JOB_ID, "Importing Song"); + await stubExportEndpoints(page); + await stubUpdateCheck(page); + await page.goto("/", { waitUntil: "domcontentloaded" }); + await page.locator(".cat-item").first().waitFor({ timeout: 20000 }); + await page.locator("#url").fill("https://www.youtube.com/watch?v=dQw4w9WgXcQ"); + await page.locator("#submit").click(); + await queue.run(); + return queue; +} + +test.describe("an import with the studio empty", () => { + test("keeps the Now Playing card empty while it runs, then opens the track", async ({ page }) => { + await seedSibling(page); + const queue = await startImport(page); + + await expect(page.locator(".app")).toHaveClass(/no-track/); + await expect(page.locator("#title")).toHaveText("Ready to import a track"); + const card = page.locator("#nowPlayingPanel"); + await expect(card).not.toContainText("Importing Song"); + await expect(card).not.toContainText(/extracting/i); + // The progress is on the library row instead. + await expect(page.locator(`.cat-item[data-id="${JOB_ID}"]`).first()).toBeVisible(); + + await queue.settle(); + await expect(page.locator(".app")).not.toHaveClass(/no-track/, { timeout: 15000 }); + await expect(page.locator("#title")).toHaveText(TRACK_TITLE, { timeout: 15000 }); + }); + + test("does not open the finished track over one the user opened meanwhile", async ({ page }) => { + await seedSibling(page); + const queue = await startImport(page); + + await page.locator(`.cat-item[data-id="${SIBLING_JOB_ID}"]`).first().click(); + await expect(page.locator("#title")).toHaveText(SIBLING_TITLE, { timeout: 15000 }); + + await queue.settle(); + await page.waitForTimeout(1500); + await expect(page.locator("#title")).toHaveText(SIBLING_TITLE); + }); + + test("does not take the studio from a Sync again import holding it", async ({ page }) => { + await seedSibling(page); + const queue = await startImport(page); + + // "Sync again" tears the player down and claims the studio by id, with no + // engine loaded. Set that claim the way importFromUrl does, through the + // same module instance the app uses. + await page.evaluate(async () => { + const state = await import("/js/state.js"); + state.setForegroundJobId("aaaaaaaaaaaa"); + }); + + await queue.settle(); + await page.waitForTimeout(1500); + await expect(page.locator(".app")).toHaveClass(/no-track/); + await expect(page.locator("#title")).not.toHaveText(TRACK_TITLE); + }); +}); diff --git a/tests/e2e/library-rows.spec.mjs b/tests/e2e/library-rows.spec.mjs index e284cc62..92c82969 100644 --- a/tests/e2e/library-rows.spec.mjs +++ b/tests/e2e/library-rows.spec.mjs @@ -89,7 +89,17 @@ test.describe("library rows", () => { await expect(page.locator(`.cat-item[data-id="${JOB_ID}"]`)).toHaveCount(0); }); + // The library asks the server where every "processing" row stands on load + // (#748), and the seeded job is done there. These rows test the label of a + // job that really is still running, so the server has to say so too. + async function serverSaysSeparating(page) { + await page.route(`**/api/jobs/${JOB_ID}`, (route) => + route.fulfill({ json: { job_id: JOB_ID, status: "separating", progress: 0.4 } }), + ); + } + test("a track still being processed says so", async ({ page }) => { + await serverSaysSeparating(page); await seedCatalogState( page, withTracks({ ...fixtureTrack(JOB_ID, "Busy"), status: "separating" }), @@ -98,6 +108,18 @@ test.describe("library rows", () => { expect(await rowText(page, JOB_ID)).toBe("Processing"); }); + test("a row left processing by a closed app takes the server's answer", async ({ page }) => { + // #748: a job that finished, failed or stopped while the app was closed + // never left the queue in front of the page, so its row said Processing + // for good. The seeded job is done on the server. + await seedCatalogState( + page, + withTracks({ ...fixtureTrack(JOB_ID, "Stale"), status: "separating" }), + ); + await open(page); + await expect.poll(() => rowText(page, JOB_ID)).not.toBe("Processing"); + }); + test("a failed import says so", async ({ page }) => { await seedCatalogState( page, @@ -108,6 +130,7 @@ test.describe("library rows", () => { }); test("the line follows the language rather than keeping the one it was saved in", async ({ page }) => { + await serverSaysSeparating(page); await page.addInitScript(() => localStorage.setItem("stemdeck.language", JSON.stringify("de")), ); diff --git a/tests/e2e/trash-queue.spec.mjs b/tests/e2e/trash-queue.spec.mjs new file mode 100644 index 00000000..3710b3a0 --- /dev/null +++ b/tests/e2e/trash-queue.spec.mjs @@ -0,0 +1,107 @@ +// Trashing an import that is still in the queue (#748), and what the Trash +// says about files on disk (#749). +// +// The queue is stubbed rather than driven: the e2e backend never runs a real +// separation, so a job that stays "running" long enough to click on cannot be +// produced against it. The server half (trash cancels the job) is covered by +// tests/test_jobs_trash.py; this covers what the page does around it. + +import { test, expect } from "@playwright/test"; + +import { fixtureTrack, seedCatalogState, stubUpdateCheck } from "./helpers.mjs"; + +const RUNNING_ID = "e2e0feedf00d"; +const RUNNING_TITLE = "Still Importing"; + +function queueFrame(running) { + return { + running: running + ? { job_id: RUNNING_ID, status: "analyzing", stage: "Analyzing audio...", progress: 0, title: RUNNING_TITLE } + : null, + queued: [], + paused: false, + max_pending_uploads: 5, + max_pending_urls: 50, + capacity_left_uploads: 5, + capacity_left_urls: 50, + }; +} + +/** + * A library holding one import that the queue reports as running, until the + * page trashes it. Counts the queue reads, so a test can tell a refresh that + * followed the trash from one that was already going to happen. + */ +async function seedRunningImport(page) { + await seedCatalogState(page, { + folders: [ + { id: "f-unsorted", name: "Unsorted", items: [RUNNING_ID], color: null }, + { id: "trash", name: "Trash", items: [], color: null }, + ], + // Its own source, or the startup sync's dedup-by-source (#542) folds it + // into the seeded fixture job and the row under test is gone. + tracks: { + [RUNNING_ID]: { + ...fixtureTrack(RUNNING_ID, RUNNING_TITLE), + status: "analyzing", + sourceUrl: "local:still-importing", + }, + }, + }); + await stubUpdateCheck(page); + + const seen = { trashed: false, queueReadsAfterTrash: 0 }; + await page.route(`**/api/jobs/${RUNNING_ID}/trash`, (route) => { + seen.trashed = true; + return route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ job_id: RUNNING_ID, trashed_at: 1 }), + }); + }); + await page.route("**/api/queue", (route) => { + if (seen.trashed) seen.queueReadsAfterTrash += 1; + return route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify(queueFrame(!seen.trashed)), + }); + }); + // The stream is held off so the only queue reads are the ones the page asks + // for, which is what the refresh test counts. + await page.route("**/api/queue/events", (route) => route.abort()); + return seen; +} + +test("a queue row offers the same trash can as the library rows", async ({ page }) => { + await seedRunningImport(page); + await page.goto("/", { waitUntil: "domcontentloaded" }); + + await page.locator(".rail-queue").click(); + const cancel = page.locator(`.queue-row[data-id="${RUNNING_ID}"] .queue-cancel`); + await expect(cancel).toBeVisible(); + // The library's trash can, not the old X (#748). + await expect(cancel.locator('polyline[points="3 6 5 6 21 6"]')).toHaveCount(1); + await expect(cancel.locator('path[d="M18 6 6 18 M6 6l12 12"]')).toHaveCount(0); +}); + +test("trashing an import that is still running refreshes the queue at once", async ({ page }) => { + const seen = await seedRunningImport(page); + await page.goto("/", { waitUntil: "domcontentloaded" }); + + // The first read has to have landed, or the page does not yet know the + // track is in the queue and has no reason to refresh it. + await page.locator(".rail-queue").click(); + await expect(page.locator(`.queue-row[data-id="${RUNNING_ID}"]`)).toBeVisible(); + await page.locator(".rail-library").click(); + + // The row's actions only take the pointer on hover (#724). + const row = page.locator(`.cat-item[data-id="${RUNNING_ID}"]`); + await row.hover(); + await row.locator(".cat-del").click(); + + await expect.poll(() => seen.queueReadsAfterTrash).toBeGreaterThan(0); + // The stream is aborted, so only the refresh the trash asked for can have + // told the page the queue is empty, which hides the queue button. + await expect(page.locator(".rail-queue")).toHaveClass(/(^| )hidden( |$)/); +}); diff --git a/tests/e2e/work-info.spec.mjs b/tests/e2e/work-info.spec.mjs index 9437fa67..a8a17610 100644 --- a/tests/e2e/work-info.spec.mjs +++ b/tests/e2e/work-info.spec.mjs @@ -15,6 +15,7 @@ import { seedCatalogState, stubAudioTags, stubExportEndpoints, + stubImportQueue, stubUpdateCheck, } from "./helpers.mjs"; @@ -323,12 +324,14 @@ test.describe("the work a soundtrack is from", () => { route.request().method() === "POST" ? route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify({ job_id: JOB_ID }) }) : route.fallback()); - await page.route(`**/api/jobs/${JOB_ID}/events`, (route) => - route.fulfill({ status: 200, contentType: "text/event-stream", body: `data: ${JSON.stringify(done)}\n\n` })); + // The import runs in the background (#747); the queue says when it is done. + const queue = await stubImportQueue(page, JOB_ID); const tagAsks = await openPage(page, { [SIBLING_JOB_ID]: fixtureTrack(SIBLING_JOB_ID, "E2E Fixture Track (again)") }); await page.locator("#url").fill("https://www.youtube.com/watch?v=6Nn8rBmhDjE"); await page.locator("#submit").click(); + await queue.run(); + await queue.settle(); await expect(page.locator("#title")).toHaveText("Popular", { timeout: 15000 }); await expect(page.locator("#np-artist")).toHaveText("Wicked", { timeout: 15000 }); diff --git a/tests/test_cancellation_reach.py b/tests/test_cancellation_reach.py index 8f0a79b8..c2099714 100644 --- a/tests/test_cancellation_reach.py +++ b/tests/test_cancellation_reach.py @@ -15,7 +15,7 @@ from app.core import process as _process from app.core.models import Job from app.core.registry import register as registry_register -from app.pipeline import runner as _runner +from app.pipeline import cancel as _cancel def _job(**kw): @@ -30,10 +30,10 @@ def _capture(job_id, proc): seen.setdefault("during", proc if proc is not None else seen.get("during")) seen["last"] = proc - monkeypatch.setattr(_runner, "set_proc", _capture) + monkeypatch.setattr(_cancel, "set_proc", _capture) job = _job() - rc, _ = _runner._run_registered_ffmpeg(job, ["sh", "-c", "exit 0"], 30) + rc = _cancel.run_registered(job, ["sh", "-c", "exit 0"], 30).returncode assert rc == 0 assert seen["during"] is not None, "cancel could not have reached this process" @@ -42,10 +42,11 @@ def _capture(job_id, proc): def test_a_failing_command_still_deregisters(tmp_path, monkeypatch): cleared = [] - monkeypatch.setattr(_runner, "set_proc", lambda job_id, proc: cleared.append(proc)) + monkeypatch.setattr(_cancel, "set_proc", lambda job_id, proc: cleared.append(proc)) job = _job() - rc, stderr = _runner._run_registered_ffmpeg(job, ["sh", "-c", "echo boom >&2; exit 3"], 30) + result = _cancel.run_registered(job, ["sh", "-c", "echo boom >&2; exit 3"], 30) + rc, stderr = result.returncode, result.stderr assert rc == 3 assert b"boom" in stderr, "stderr must still be captured for the error message" @@ -168,3 +169,85 @@ def test_cancelling_a_plain_done_job_still_does_nothing(monkeypatch): assert not proc.terminated assert job.cancel_requested is False + + +# ─── the cancelled job's folder (#749) ─── + + +async def test_a_locked_file_does_not_leave_a_cancelled_jobs_folder_behind(tmp_path, monkeypatch): + """Right after a cancel terminates ffmpeg or the demucs worker, Windows can + still hold one of its files. One attempt left the folder on disk for good, + the orphan #749 reported; the removal is retried like delete_job's.""" + import shutil + + from app.core.models import JobCancelled + from app.pipeline import collect as _collect + from app.pipeline import runner as _runner + + job = _job(cancel_requested=True) + registry_register(job) + job_dir = tmp_path / job.id + (job_dir / "stems").mkdir(parents=True) + (job_dir / "stems" / "vocals.wav").write_bytes(b"RIFF") + + real_rmtree = shutil.rmtree + attempts = [] + + def locked_once(path, *a, **k): + attempts.append(path) + if len(attempts) == 1: + raise PermissionError(32, "The process cannot access the file", str(path)) + return real_rmtree(path, *a, **k) + + monkeypatch.setattr(_collect.shutil, "rmtree", locked_once) + + def cancelled(_job, _job_dir): + raise JobCancelled() + + await _runner._run_async(job, job_dir, tmp_path, cancelled) + + assert job.status == "cancelled" + assert len(attempts) == 2, "the lock must be retried, not given up on" + assert not job_dir.exists(), "the cancelled job's folder stayed on disk" + + +# ─── a job stopped from the Trash (#748) ─── + + +async def test_a_run_stopped_from_the_trash_keeps_its_files(tmp_path): + """Only emptying the Trash deletes files, so Restore and Extract have the + source to work from.""" + from app.core.models import JobCancelled + from app.pipeline import runner as _runner + + job = _job(cancel_requested=True, stop_requested=True) + registry_register(job) + job_dir = tmp_path / job.id + job_dir.mkdir() + (job_dir / "source.wav").write_bytes(b"RIFF") + + def cancelled(_job, _job_dir): + raise JobCancelled() + + await _runner._run_async(job, job_dir, tmp_path, cancelled) + + assert job.status == "stopped" + assert (job_dir / "source.wav").is_file() + + +async def test_a_run_that_finishes_despite_a_late_trash_clears_the_flags(tmp_path): + """The trash arrived after the last cancel check, so the job finished. Left + set, the flags made every later vocal split or section run on it cancel + itself at once.""" + from app.pipeline import runner as _runner + + job = _job(cancel_requested=True, stop_requested=True) + registry_register(job) + job_dir = tmp_path / job.id + job_dir.mkdir() + + await _runner._run_async(job, job_dir, tmp_path, lambda _job, _job_dir: None) + + assert job.status == "done" + assert job.cancel_requested is False + assert job.stop_requested is False diff --git a/tests/test_jobs_trash.py b/tests/test_jobs_trash.py index 5b87793b..141fee67 100644 --- a/tests/test_jobs_trash.py +++ b/tests/test_jobs_trash.py @@ -14,6 +14,8 @@ from __future__ import annotations +from pathlib import Path + import pytest from fastapi.testclient import TestClient @@ -147,3 +149,149 @@ def test_a_registry_written_before_this_existed_still_loads() -> None: record = Job(id="666666666666", status="done", title="Old").to_record() del record["trashed_at"] assert Job.from_record(record).trashed_at is None + + +# ── trashing a job that has not finished (#748) ───────────────────────────── +# +# Trashing used to leave an unfinished import running, hidden: the queue kept +# working on a song the user had thrown away. The server now stops it, so the +# phone and the desktop both get it. Its files stay: only emptying the Trash +# deletes anything. + + +class _FakeProc: + """Stands in for the registered subprocess: records the terminate.""" + + def __init__(self) -> None: + self.terminated = False + + def poll(self): + return 0 if self.terminated else None + + def terminate(self) -> None: + self.terminated = True + + +def _queued_upload(job_id: str) -> tuple[Job, Path]: + from app.pipeline import jobqueue + + job = Job(id=job_id, status="queued", title="Waiting", source_url="local:Waiting") + register(job) + jobqueue.enqueue(job.id, autostart=False) + job_dir = jobqueue.JOBS_DIR / job.id + job_dir.mkdir(parents=True) + (job_dir / "source.wav").write_bytes(b"RIFF") + return job, job_dir + + +def test_trashing_a_queued_job_stops_it_and_keeps_its_files(client: TestClient) -> None: + from app.pipeline import jobqueue + + job, job_dir = _queued_upload("888888888888") + + assert client.post(f"/api/jobs/{job.id}/trash").status_code == 200 + + assert job.status == "stopped" + assert jobqueue.snapshot() == (None, []), "the queue would still run it" + assert (job_dir / "source.wav").is_file(), "only emptying the Trash deletes files" + + +def test_a_plain_cancel_still_removes_a_queued_jobs_files(client: TestClient) -> None: + job, job_dir = _queued_upload("888888888887") + + assert client.post(f"/api/jobs/{job.id}/cancel").status_code == 200 + + assert job.status == "cancelled" + assert not job_dir.exists(), "a waiting upload's source must not stay on disk" + + +def test_emptying_the_trash_deletes_a_stopped_job(client: TestClient) -> None: + job, job_dir = _queued_upload("888888888886") + client.post(f"/api/jobs/{job.id}/trash") + + assert client.delete(f"/api/jobs/{job.id}").status_code == 200 + + assert not job_dir.exists() + + +def test_extract_queues_a_restored_stopped_job_again(client: TestClient) -> None: + from app.core.config import DEMUCS_MODEL + from app.pipeline import jobqueue + + job, job_dir = _queued_upload("888888888885") + client.post(f"/api/jobs/{job.id}/trash") + (job_dir / DEMUCS_MODEL).mkdir() # partial output from the stopped run + client.post(f"/api/jobs/{job.id}/restore") + + res = client.post(f"/api/jobs/{job.id}/extract") + + assert res.status_code == 200 + assert res.json()["status"] == "queued" + assert job.cancel_requested is False and job.stop_requested is False + assert jobqueue.snapshot() == (None, [job.id]) + assert (job_dir / "source.wav").is_file() + assert not (job_dir / DEMUCS_MODEL).exists(), "collect() would read stale stems" + + +def test_restore_alone_does_not_queue_a_stopped_job(client: TestClient) -> None: + from app.pipeline import jobqueue + + job, _ = _queued_upload("888888888884") + client.post(f"/api/jobs/{job.id}/trash") + + assert client.post(f"/api/jobs/{job.id}/restore").status_code == 200 + + assert job.status == "stopped" + assert jobqueue.snapshot() == (None, []) + + +def test_extract_refuses_a_job_still_in_the_trash(client: TestClient) -> None: + job, _ = _queued_upload("888888888883") + client.post(f"/api/jobs/{job.id}/trash") + + assert client.post(f"/api/jobs/{job.id}/extract").status_code == 409 + assert job.status == "stopped" + + +def test_extract_refuses_a_job_that_is_not_stopped(client: TestClient) -> None: + job = _done("888888888882", "Finished") + assert client.post(f"/api/jobs/{job.id}/extract").status_code == 409 + assert job.status == "done" + + +def test_extract_rejects_a_malformed_or_unknown_id(client: TestClient) -> None: + # A traversal normalises away before routing (405), as for trash above. + for bad in ("../../etc/passwd", "not a job id", "%2e%2e%2f"): + resp = client.post(f"/api/jobs/{bad}/extract") + assert 400 <= resp.status_code < 500, (bad, resp.status_code) + assert client.post("/api/jobs/aaaaaaaaaaaa/extract").status_code == 404 + + +def test_trashing_the_running_job_stops_its_process(client: TestClient) -> None: + from app.core.registry import set_proc + from app.pipeline import jobqueue + + job = Job(id="999999999990", status="analyzing", title="Running") + register(job) + jobqueue._set_running(job.id) + proc = _FakeProc() + set_proc(job.id, proc) + try: + assert client.post(f"/api/jobs/{job.id}/trash").status_code == 200 + finally: + set_proc(job.id, None) + jobqueue._set_running(None) + + assert job.cancel_requested is True + assert job.stop_requested is True, "it would settle as cancelled and lose its files" + assert proc.terminated, "the pipeline would have kept going in the Trash" + # Still in the Trash: the runner, not the endpoint, finalises a running job. + assert job.trashed_at is not None + + +def test_trashing_a_finished_job_cancels_nothing(client: TestClient) -> None: + """A done job's files are what Restore brings back.""" + job = _done("999999999991", "Finished") + client.post(f"/api/jobs/{job.id}/trash") + assert job.status == "done" + assert job.cancel_requested is False diff --git a/tests/test_pipeline_analyze.py b/tests/test_pipeline_analyze.py index 9b62b3da..75265e86 100644 --- a/tests/test_pipeline_analyze.py +++ b/tests/test_pipeline_analyze.py @@ -446,3 +446,79 @@ def test_a_silent_track_yields_no_key(job, jobs_dir): bpm, key = az.analyze(job, src) assert key is None or key.split()[0] in PITCHES assert bpm is None or bpm > 0 + + +# ── cancel (#748) ──────────────────────────────────────────────────── + + +def test_the_analyze_decode_is_registered_so_cancel_can_stop_it(job, jobs_dir, monkeypatch): + """subprocess.run() could not be reached by a cancel: a song trashed while + analysing stayed "Analyzing" until the decode and every step after it + finished (85 s in the report).""" + import numpy as np + + src = jobs_dir / "x.wav" + src.write_bytes(b"RIFF") + calls = [] + + def fake_run_registered(job_arg, cmd, timeout, *, capture_stdout=False): + calls.append((job_arg, capture_stdout)) + samples = np.zeros(22050, dtype=np.float32).tobytes() + return subprocess.CompletedProcess(cmd, 0, samples, b"") + + monkeypatch.setattr(az, "run_registered", fake_run_registered) + + result = az._load_audio_ffmpeg(src, job=job) + + assert result is not None + assert calls == [(job, True)] + + +def test_a_terminated_decode_reads_as_a_failure(job, jobs_dir, monkeypatch): + src = jobs_dir / "x.wav" + src.write_bytes(b"RIFF") + monkeypatch.setattr( + az, + "run_registered", + lambda _j, cmd, _t, **_k: subprocess.CompletedProcess(cmd, 1, b"", b""), + ) + assert az._load_audio_ffmpeg(src, job=job) is None + + +def test_a_cancel_during_the_decode_raises_rather_than_skipping(job, jobs_dir, monkeypatch): + """The cancel kills ffmpeg, which looks like a failed decode. Treated as + one, the job went on to separate a song the user had thrown away.""" + from app.core.models import JobCancelled + + pytest.importorskip("librosa") + src = jobs_dir / "x.wav" + src.write_bytes(b"RIFF") + + def cancelled_decode(job_arg, cmd, _t, **_k): + job_arg.cancel_requested = True + return subprocess.CompletedProcess(cmd, 1, b"", b"") + + monkeypatch.setattr(az, "run_registered", cancelled_decode) + + with pytest.raises(JobCancelled): + az.analyze(job, src) + + +def test_a_cancel_between_steps_is_not_swallowed(job, jobs_dir): + """The librosa steps cannot be interrupted, so the flag is checked between + them, and the stage's catch-all must let the cancel through.""" + from app.core.models import JobCancelled + + skip_without_ffmpeg() + librosa = pytest.importorskip("librosa") + src = _click_wav(jobs_dir / "c.wav", bpm=120.0, seconds=4.0) + real_hpss = librosa.effects.hpss + + def hpss_then_cancel(y, *a, **k): + job.cancel_requested = True + return real_hpss(y, *a, **k) + + with patch.object(librosa.effects, "hpss", hpss_then_cancel), pytest.raises(JobCancelled): + az.analyze(job, src) + + assert job.bpm is None, "a cancelled analysis must not write its results" diff --git a/tests/test_registry_persistence.py b/tests/test_registry_persistence.py index fe4468d3..a37cded3 100644 --- a/tests/test_registry_persistence.py +++ b/tests/test_registry_persistence.py @@ -414,3 +414,50 @@ def test_records_without_a_position_still_restore_oldest_first(tmp_path: Path): restore_registry(tmp_path) assert _registry.take_pending_resume() == ["abcdef0000e2", "abcdef0000e1"] + + +# ── a job trashed before the restart (#748) ────────────────────────────────── + + +def test_a_trashed_interrupted_job_is_not_put_back_in_the_queue(tmp_path: Path): + """It came back as a Paused row offering to Start a song in the Trash.""" + job = Job(id="abcdef0000f1", status="analyzing", title="Binned", trashed_at=1.0) + _jobs[job.id] = job + persist_registry(tmp_path) + (tmp_path / job.id).mkdir() + (tmp_path / job.id / "source.wav").write_bytes(b"RIFF") + _jobs.clear() + + restore_registry(tmp_path) + + assert _registry.take_pending_resume() == [] + assert _jobs[job.id].status == "stopped" + assert (tmp_path / job.id / "source.wav").is_file(), "only emptying the Trash deletes files" + + +def test_a_stopped_job_survives_a_restart_in_the_trash(tmp_path: Path): + """Not persisted, it would vanish from the Trash with its files orphaned.""" + job = Job(id="abcdef0000f3", status="stopped", trashed_at=1.0) # no title yet + _jobs[job.id] = job + persist_registry(tmp_path) + _jobs.clear() + + restore_registry(tmp_path) + + assert _jobs[job.id].status == "stopped" + assert _jobs[job.id].trashed_at == 1.0 + assert _registry.take_pending_resume() == [], "only Extract may queue it again" + + +def test_a_trashed_job_that_finished_before_the_crash_stays_in_the_trash(tmp_path: Path): + job = Job(id="abcdef0000f2", status="separating", title="Binned", trashed_at=1.0) + _jobs[job.id] = job + persist_registry(tmp_path) + _stems_dir(tmp_path, job.id) + _jobs.clear() + + restore_registry(tmp_path) + + assert _jobs[job.id].status == "done" + assert _jobs[job.id].trashed_at == 1.0, "a song the user binned would reappear" + assert _registry.take_pending_resume() == [] diff --git a/tests/test_video_status.py b/tests/test_video_status.py index d89a71e5..5e1177f6 100644 --- a/tests/test_video_status.py +++ b/tests/test_video_status.py @@ -97,16 +97,16 @@ def _run_local_extract(tmp_path: Path, returncode: int, raises=None) -> Job: source = job_dir / "in.mp4" source.write_bytes(b"x") - # Stubs _run_registered_ffmpeg rather than subprocess.run: the extract goes + # Stubs run_registered rather than subprocess.run: the extract goes # through Popen + set_proc now, so cancel can reach it (#519). def fake_run(job_arg, cmd, timeout): if raises is not None: raise raises if returncode == 0: Path(cmd[-1]).write_bytes(b"fake mp4 payload") - return returncode, b"" + return subprocess.CompletedProcess(cmd, returncode, b"", b"") - with patch.object(runner_mod, "_run_registered_ffmpeg", fake_run): + with patch.object(runner_mod, "run_registered", fake_run): runner_mod._extract_video_track(job, source, job_dir) return job