diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index 5d79a4a6..ad601cf6 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -16,6 +16,7 @@ from app.config import settings from app.downloads import adapters from app.downloads import edit as editmod from app.downloads import formats, quota, storage +from app.downloads.status import HOLDING from app.models import Channel, DownloadJob, MediaAsset, Video @@ -238,7 +239,7 @@ def release_asset(db: Session, job: DownloadJob) -> None: +1 it, and deleting it under that would FK-null the new job's asset (a lost update on ref_count). A ready asset is still freed at ref==0 (its file is the point); an orphaned errored row is cheap (no file) and gets reused+reset by the next enqueue, or reclaimed by a later GC pass.""" - if not job.asset_id or job.status not in ("queued", "running", "paused", "done", "error"): + if not job.asset_id or job.status not in HOLDING: return asset = db.get(MediaAsset, job.asset_id) if asset is None: diff --git a/backend/app/downloads/status.py b/backend/app/downloads/status.py new file mode 100644 index 00000000..7ee2a05f --- /dev/null +++ b/backend/app/downloads/status.py @@ -0,0 +1,38 @@ +"""Canonical download-job status values — the ONE source of truth for the lifecycle literals. + +A job's status is written from several places: the worker (claim → running, terminal done/error, +requeue), the routes (pause/cancel/redownload/delete), the service (enqueue), and the GC. Keeping the +literals here is a data-safety measure, not tidiness. A British/American drift once silently broke +cancellation: the supervisor compared the live status against ``"cancelled"`` while the cancel route +wrote ``"canceled"`` (models.py), so the supervisor's "a route stopped it → kill the child, free the +asset" branch was DEAD. A canceled job's child then ran to completion and wrote ``"done"`` +unconditionally over the route's ``"canceled"``; because ``done`` counts as a holding state but +``canceled`` does not (see ``service.release_asset``), a later delete released the asset a SECOND time, +dropping ``ref_count`` below the live holder count and deleting a file still in use. Import these names +instead of typing the strings. +""" + +from enum import StrEnum + + +class JobStatus(StrEnum): + QUEUED = "queued" + RUNNING = "running" + PAUSED = "paused" + DONE = "done" + ERROR = "error" + CANCELED = "canceled" + + +# A job "holds" its media asset (the +1 on ``ref_count`` from enqueue) while in any of these states; +# ``release_asset`` on a status OUTSIDE this set is a no-op. ``CANCELED`` is deliberately excluded: the +# cancel route releases once while the job is still holding, then sets ``CANCELED``, so a later delete +# does not double-release. This invariant only holds while a finishing child cannot flip ``CANCELED`` +# back into the set — hence the worker's terminal writes are conditional on the row still being RUNNING. +HOLDING = frozenset( + {JobStatus.QUEUED, JobStatus.RUNNING, JobStatus.PAUSED, JobStatus.DONE, JobStatus.ERROR} +) + +# States a route sets to STOP a running job. The worker's supervisor must step in (kill the child, free +# the asset, leave the route's status intact) when it observes one of these on a job it is running. +STOPPED_BY_ROUTE = frozenset({JobStatus.PAUSED, JobStatus.CANCELED}) diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index e1fe1503..2b1c08c9 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -22,6 +22,7 @@ from app.db import get_db from app.downloads import edit as editmod from app.downloads import links as linksmod from app.downloads import formats, quota, service, storage +from app.downloads.status import JobStatus from app.models import ( DownloadJob, DownloadLink, @@ -544,9 +545,9 @@ def cancel_download( job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: job = _own_job(db, user, job_id) - if job.status not in ("done", "canceled"): + if job.status not in (JobStatus.DONE, JobStatus.CANCELED): service.release_asset(db, job) # release while the job still counts as holding - job.status = "canceled" + job.status = JobStatus.CANCELED db.commit() return _serialize_job(db, job) diff --git a/backend/app/worker.py b/backend/app/worker.py index 0749bd42..d7624203 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -36,6 +36,7 @@ from app.db import SessionLocal from app.downloads import edit as editmod from app.downloads import errors as dlerrors from app.downloads import adapters, formats, quota, service, storage +from app.downloads.status import STOPPED_BY_ROUTE, JobStatus from app.models import DownloadJob, MediaAsset from app.notifications import create_notification from app.titles import normalize_title @@ -359,7 +360,7 @@ def _run_supervised(job_id: int, asset_id: int) -> None: except subprocess.TimeoutExpired: pass status, stale = _job_liveness(job_id) - if status is None or status in ("paused", "cancelled"): + if status is None or status in STOPPED_BY_ROUTE: reason = "cancelled" # a route stopped it (or the row vanished) break if status == "running": @@ -481,8 +482,11 @@ def _finish_from_cache(job_id: int, asset_id: int) -> None: with SessionLocal() as db: job = db.get(DownloadJob, job_id) asset = db.get(MediaAsset, asset_id) - if job and asset: - job.status = "done" + # Only finalize a job the route hasn't stopped: the flip is conditional on the row still being + # RUNNING so a concurrent cancel/pause is never resurrected to `done` (which would break the + # no-double-release invariant in service.release_asset — see downloads/status.py). + if job and asset and job.status == JobStatus.RUNNING: + job.status = JobStatus.DONE job.progress = 100 job.phase = None if not job.display_name: @@ -868,8 +872,12 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe asset.last_access_at = datetime.now(timezone.utc) asset.expires_at = expires job = db.get(DownloadJob, job_id) - if job: - job.status = "done" + # Conditional terminal write: only finalize while the row is still RUNNING, so a child that + # finishes AFTER a route cancel/pause (narrow window past the cooperative checkpoints) can't + # overwrite `canceled`/`paused` with `done` and re-open the double-release class fixed here. + finalized = bool(job and job.status == JobStatus.RUNNING) + if finalized: + job.status = JobStatus.DONE job.progress = 100 job.phase = None job.speed_bps = None @@ -880,7 +888,8 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe # Capture the owner before commit expires the row (avoids a refresh SELECT just to notify). owner_id = job.user_id if job else None db.commit() - _notify_download_done(db, owner_id, meta.title, job_id, kind="download") + if finalized: + _notify_download_done(db, owner_id, meta.title, job_id, kind="download") log.info("job %s done: %s", job_id, rel) except (_Aborted, editmod.EditAborted): @@ -896,8 +905,10 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe asset.status = "error" asset.error = msg job = db.get(DownloadJob, job_id) - if job: - job.status = "error" + # Don't overwrite a route-set cancel/pause with `error` (error also holds the asset, so it + # would re-open the double-release class): only fail a job that is still RUNNING. + if job and job.status == JobStatus.RUNNING: + job.status = JobStatus.ERROR job.error = msg job.error_code = dlerrors.classify(msg) job.phase = None @@ -928,8 +939,10 @@ def _fail_edit(job_id: int, asset_id: int, msg: str) -> None: asset.status = "error" asset.error = msg job = db.get(DownloadJob, job_id) - if job: - job.status = "error" + # Only fail a job that is still RUNNING — never overwrite a route-set cancel/pause (see the + # download error path). + if job and job.status == JobStatus.RUNNING: + job.status = JobStatus.ERROR job.error = msg job.error_code = dlerrors.classify(msg) job.phase = None @@ -1008,15 +1021,18 @@ def _process_edit(job_id: int, asset_id: int) -> None: asset.last_access_at = datetime.now(timezone.utc) asset.expires_at = expires job = db.get(DownloadJob, job_id) - if job: - job.status = "done" + # Conditional terminal write (see the download path): don't resurrect a route-stopped job. + finalized = bool(job and job.status == JobStatus.RUNNING) + if finalized: + job.status = JobStatus.DONE job.progress = 100 job.phase = None # Capture owner + title before commit expires the rows (avoids a refresh SELECT). owner_id = job.user_id if job else None dl_title = asset.title if asset else None db.commit() - _notify_download_done(db, owner_id, dl_title, job_id, kind="edit") + if finalized: + _notify_download_done(db, owner_id, dl_title, job_id, kind="edit") log.info("edit job %s done: %s", job_id, rel) except editmod.EditAborted: diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py index e4b66006..a01b85c2 100644 --- a/backend/tests/test_worker_supervisor.py +++ b/backend/tests/test_worker_supervisor.py @@ -88,7 +88,7 @@ def test_no_progress_stall_kills_and_fails_the_job(monkeypatch, rec): def test_route_pause_kills_but_keeps_the_route_status(monkeypatch, rec): - # A route flipped the job to 'paused'/'cancelled'. Kill + free the asset, but do NOT overwrite the + # A route flipped the job to 'paused'. Kill + free the asset, but do NOT overwrite the # status with an error and do NOT requeue. _spawn(monkeypatch, _FakeProc(["timeout"])) _liveness(monkeypatch, ("paused", 1.0)) @@ -98,6 +98,19 @@ def test_route_pause_kills_but_keeps_the_route_status(monkeypatch, rec): assert "set_job" not in names and "requeue" not in names +def test_route_cancel_kills_but_keeps_the_route_status(monkeypatch, rec): + # B-sub-1 regression: the cancel ROUTE writes the literal 'canceled' (models.py), but the supervisor + # used to compare against 'cancelled' — so this branch was DEAD and a canceled job's child ran on to + # completion, overwriting 'canceled' with 'done' and double-releasing its asset. The supervisor MUST + # treat the route's actual literal as a stop: kill + free the asset, no error write, no requeue. + _spawn(monkeypatch, _FakeProc(["timeout"])) + _liveness(monkeypatch, ("canceled", 1.0)) + worker._run_supervised(1, 99) + names = _names(rec) + assert "stop_child" in names and "reset_asset" in names + assert "set_job" not in names and "requeue" not in names + + def test_wall_clock_cap_kills_even_while_progressing(monkeypatch, rec): monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0) _spawn(monkeypatch, _FakeProc(["timeout"]))