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..d015e4a0 100644 --- a/static/css/daw.css +++ b/static/css/daw.css @@ -578,8 +578,9 @@ input, textarea { font-family: inherit; } .rail-btn > span { white-space: nowrap; max-width: 56px; overflow: hidden; text-overflow: ellipsis; } /* Queue rail button. Only present while something is importing, so the rail - stays as it is today for anyone not using the queue. */ -.rail-queue { position: relative; } + stays as it is today for anyone not using the queue. The Trash button wears + the same badge, with how many songs are in it (#749). */ +.rail-queue, .rail-trash { position: relative; } .rail-badge { position: absolute; top: 2px; right: 2px; min-width: 15px; height: 15px; padding: 0 4px; @@ -929,6 +930,42 @@ 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; } + +/* Trash rows look like library rows. On hover the second line becomes the + hint, and a hold fills the row from the left over --hold-ms (set from + TRASH_HOLD_MS in catalog.js), then restores. */ +.cat-item.in-trash { user-select: none; -webkit-user-select: none; overflow: hidden; } +.cat-item.in-trash .trash-hint { display: none; color: var(--fg-2); } +.cat-item.in-trash:hover .trash-hint, +.cat-item.in-trash:focus-visible .trash-hint, +.cat-item.in-trash.holding .trash-hint { display: block; overflow: hidden; text-overflow: ellipsis; } +.cat-item.in-trash:hover .cat-sub:not(.trash-hint), +.cat-item.in-trash:focus-visible .cat-sub:not(.trash-hint), +.cat-item.in-trash.holding .cat-sub:not(.trash-hint) { display: none; } +.trash-hold { position: absolute; inset: 0; pointer-events: none; } +.trash-hold::before { + content: ""; position: absolute; inset: 0; + background: color-mix(in srgb, var(--accent) 18%, transparent); + transform: scaleX(0); transform-origin: left center; +} +.cat-item.in-trash.holding .trash-hold::before { + transform: scaleX(1); + transition: transform var(--hold-ms, 1000ms) linear; +} +.cat-item.in-trash.selected { + background: color-mix(in srgb, var(--accent) 10%, transparent); + border-color: color-mix(in srgb, var(--accent) 45%, transparent); +} +.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; @@ -5632,3 +5669,39 @@ input, textarea { font-family: inherit; } box-shadow: 0 12px 32px rgba(0, 0, 0, 0.42); } .stem-overflow.open { display: flex; } + +/* Help mode. The Guide button in the rail, below Trash; while it is pressed, + a ring is drawn around what can be seen of the control being explained, + and its box sits beside it. The ring is a layer of its own, so nothing in + the page moves and no panel clips it. */ +.rail-guide[aria-pressed="true"] { color: var(--accent); background: var(--panel); } +/* Only the control being explained is marked, so the screen stays calm; the + pill at the bottom says help mode is on. */ +.help-mode .help-target { cursor: help; } +.help-ring { + position: fixed; z-index: 9998; pointer-events: none; + border: 2px solid var(--accent); border-radius: var(--radius-sm); + box-shadow: 0 0 0 4px color-mix(in srgb, var(--accent) 16%, transparent); +} +.help-pill { + position: fixed; z-index: 9999; left: 50%; bottom: 18px; transform: translateX(-50%); + max-width: calc(100vw - 32px); + background: var(--panel); border: 1px solid var(--accent); border-radius: 999px; + padding: 7px 16px; box-shadow: var(--shadow-panel); + color: var(--fg); font: 500 12.5px/1.4 var(--font-sans); text-align: center; + pointer-events: none; +} +.help-tip { + position: fixed; z-index: 10000; + width: max-content; max-width: min(300px, calc(100vw - 16px)); + background: var(--panel); border: 1px solid var(--accent); border-radius: var(--radius); + padding: 10px 12px; box-shadow: var(--shadow-panel); + color: var(--fg-2); font: 400 13px/1.45 var(--font-sans); + pointer-events: none; +} +.help-tip b { display: block; margin-bottom: 2px; color: var(--fg); font-size: 13.5px; font-weight: 700; } +.help-tip.long { max-width: min(560px, calc(100vw - 16px)); columns: 2 230px; column-gap: 20px; font-size: 12px; line-height: 1.4; } +.help-tip.long b { column-span: all; margin-bottom: 6px; } +.help-tip-h { break-after: avoid; margin-top: 8px; color: var(--accent); font-size: 10px; font-weight: 700; letter-spacing: 0.1em; text-transform: uppercase; } +.help-tip-row { break-inside: avoid; margin-top: 3px; } +.help-tip-row span { color: var(--fg); font-weight: 600; } diff --git a/static/index.html b/static/index.html index 7baf25c0..c863c3af 100644 --- a/static/index.html +++ b/static/index.html @@ -323,11 +323,21 @@ Queue - + +