fix(downloads): revive the worker cancel branch and guard terminal writes (B-sub-1)
The cancel route writes status 'canceled' but the supervisor compared against 'cancelled', so its "a route stopped it -> kill the child, free the asset" branch was dead code. A canceled job's child ran to completion and wrote 'done' unconditionally over the route's 'canceled'; since 'done' holds the asset but 'canceled' does not, a later delete released the asset a second time, dropping ref_count below the live holder count and deleting a file still in use. Add app/downloads/status.py as the single source of truth for the lifecycle literals (JobStatus StrEnum + the HOLDING / STOPPED_BY_ROUTE sets) and use it at the safety-critical sites: the supervisor stop check, the route cancel, and release_asset's holding set. Make the worker's terminal done/error writes conditional on the row still being 'running' so a child finishing just after a route cancel/pause can no longer resurrect it. Add the missing 'canceled' supervisor regression test. (The full literal sweep across the downloads surface is left for the E3 downloads owner-unification; this is the data-safety subset.)
This commit is contained in:
@@ -16,6 +16,7 @@ from app.config import settings
|
|||||||
from app.downloads import adapters
|
from app.downloads import adapters
|
||||||
from app.downloads import edit as editmod
|
from app.downloads import edit as editmod
|
||||||
from app.downloads import formats, quota, storage
|
from app.downloads import formats, quota, storage
|
||||||
|
from app.downloads.status import HOLDING
|
||||||
from app.models import Channel, DownloadJob, MediaAsset, Video
|
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).
|
+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
|
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."""
|
(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
|
return
|
||||||
asset = db.get(MediaAsset, job.asset_id)
|
asset = db.get(MediaAsset, job.asset_id)
|
||||||
if asset is None:
|
if asset is None:
|
||||||
|
|||||||
@@ -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})
|
||||||
@@ -22,6 +22,7 @@ from app.db import get_db
|
|||||||
from app.downloads import edit as editmod
|
from app.downloads import edit as editmod
|
||||||
from app.downloads import links as linksmod
|
from app.downloads import links as linksmod
|
||||||
from app.downloads import formats, quota, service, storage
|
from app.downloads import formats, quota, service, storage
|
||||||
|
from app.downloads.status import JobStatus
|
||||||
from app.models import (
|
from app.models import (
|
||||||
DownloadJob,
|
DownloadJob,
|
||||||
DownloadLink,
|
DownloadLink,
|
||||||
@@ -544,9 +545,9 @@ def cancel_download(
|
|||||||
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
||||||
) -> dict:
|
) -> dict:
|
||||||
job = _own_job(db, user, job_id)
|
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
|
service.release_asset(db, job) # release while the job still counts as holding
|
||||||
job.status = "canceled"
|
job.status = JobStatus.CANCELED
|
||||||
db.commit()
|
db.commit()
|
||||||
return _serialize_job(db, job)
|
return _serialize_job(db, job)
|
||||||
|
|
||||||
|
|||||||
+29
-13
@@ -36,6 +36,7 @@ from app.db import SessionLocal
|
|||||||
from app.downloads import edit as editmod
|
from app.downloads import edit as editmod
|
||||||
from app.downloads import errors as dlerrors
|
from app.downloads import errors as dlerrors
|
||||||
from app.downloads import adapters, formats, quota, service, storage
|
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.models import DownloadJob, MediaAsset
|
||||||
from app.notifications import create_notification
|
from app.notifications import create_notification
|
||||||
from app.titles import normalize_title
|
from app.titles import normalize_title
|
||||||
@@ -359,7 +360,7 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
|
|||||||
except subprocess.TimeoutExpired:
|
except subprocess.TimeoutExpired:
|
||||||
pass
|
pass
|
||||||
status, stale = _job_liveness(job_id)
|
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)
|
reason = "cancelled" # a route stopped it (or the row vanished)
|
||||||
break
|
break
|
||||||
if status == "running":
|
if status == "running":
|
||||||
@@ -481,8 +482,11 @@ def _finish_from_cache(job_id: int, asset_id: int) -> None:
|
|||||||
with SessionLocal() as db:
|
with SessionLocal() as db:
|
||||||
job = db.get(DownloadJob, job_id)
|
job = db.get(DownloadJob, job_id)
|
||||||
asset = db.get(MediaAsset, asset_id)
|
asset = db.get(MediaAsset, asset_id)
|
||||||
if job and asset:
|
# Only finalize a job the route hasn't stopped: the flip is conditional on the row still being
|
||||||
job.status = "done"
|
# 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.progress = 100
|
||||||
job.phase = None
|
job.phase = None
|
||||||
if not job.display_name:
|
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.last_access_at = datetime.now(timezone.utc)
|
||||||
asset.expires_at = expires
|
asset.expires_at = expires
|
||||||
job = db.get(DownloadJob, job_id)
|
job = db.get(DownloadJob, job_id)
|
||||||
if job:
|
# Conditional terminal write: only finalize while the row is still RUNNING, so a child that
|
||||||
job.status = "done"
|
# 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.progress = 100
|
||||||
job.phase = None
|
job.phase = None
|
||||||
job.speed_bps = 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).
|
# Capture the owner before commit expires the row (avoids a refresh SELECT just to notify).
|
||||||
owner_id = job.user_id if job else None
|
owner_id = job.user_id if job else None
|
||||||
db.commit()
|
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)
|
log.info("job %s done: %s", job_id, rel)
|
||||||
|
|
||||||
except (_Aborted, editmod.EditAborted):
|
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.status = "error"
|
||||||
asset.error = msg
|
asset.error = msg
|
||||||
job = db.get(DownloadJob, job_id)
|
job = db.get(DownloadJob, job_id)
|
||||||
if job:
|
# Don't overwrite a route-set cancel/pause with `error` (error also holds the asset, so it
|
||||||
job.status = "error"
|
# 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 = msg
|
||||||
job.error_code = dlerrors.classify(msg)
|
job.error_code = dlerrors.classify(msg)
|
||||||
job.phase = None
|
job.phase = None
|
||||||
@@ -928,8 +939,10 @@ def _fail_edit(job_id: int, asset_id: int, msg: str) -> None:
|
|||||||
asset.status = "error"
|
asset.status = "error"
|
||||||
asset.error = msg
|
asset.error = msg
|
||||||
job = db.get(DownloadJob, job_id)
|
job = db.get(DownloadJob, job_id)
|
||||||
if job:
|
# Only fail a job that is still RUNNING — never overwrite a route-set cancel/pause (see the
|
||||||
job.status = "error"
|
# download error path).
|
||||||
|
if job and job.status == JobStatus.RUNNING:
|
||||||
|
job.status = JobStatus.ERROR
|
||||||
job.error = msg
|
job.error = msg
|
||||||
job.error_code = dlerrors.classify(msg)
|
job.error_code = dlerrors.classify(msg)
|
||||||
job.phase = None
|
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.last_access_at = datetime.now(timezone.utc)
|
||||||
asset.expires_at = expires
|
asset.expires_at = expires
|
||||||
job = db.get(DownloadJob, job_id)
|
job = db.get(DownloadJob, job_id)
|
||||||
if job:
|
# Conditional terminal write (see the download path): don't resurrect a route-stopped job.
|
||||||
job.status = "done"
|
finalized = bool(job and job.status == JobStatus.RUNNING)
|
||||||
|
if finalized:
|
||||||
|
job.status = JobStatus.DONE
|
||||||
job.progress = 100
|
job.progress = 100
|
||||||
job.phase = None
|
job.phase = None
|
||||||
# Capture owner + title before commit expires the rows (avoids a refresh SELECT).
|
# Capture owner + title before commit expires the rows (avoids a refresh SELECT).
|
||||||
owner_id = job.user_id if job else None
|
owner_id = job.user_id if job else None
|
||||||
dl_title = asset.title if asset else None
|
dl_title = asset.title if asset else None
|
||||||
db.commit()
|
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)
|
log.info("edit job %s done: %s", job_id, rel)
|
||||||
|
|
||||||
except editmod.EditAborted:
|
except editmod.EditAborted:
|
||||||
|
|||||||
@@ -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):
|
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.
|
# status with an error and do NOT requeue.
|
||||||
_spawn(monkeypatch, _FakeProc(["timeout"]))
|
_spawn(monkeypatch, _FakeProc(["timeout"]))
|
||||||
_liveness(monkeypatch, ("paused", 1.0))
|
_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
|
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):
|
def test_wall_clock_cap_kills_even_while_progressing(monkeypatch, rec):
|
||||||
monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0)
|
monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0)
|
||||||
_spawn(monkeypatch, _FakeProc(["timeout"]))
|
_spawn(monkeypatch, _FakeProc(["timeout"]))
|
||||||
|
|||||||
Reference in New Issue
Block a user