From 4ce0a702f481908a352aa796cee7ed922beed66a Mon Sep 17 00:00:00 2001 From: npeter83 Date: Tue, 18 Aug 2026 00:58:13 +0200 Subject: [PATCH] refactor(downloads): route worker terminal writes through an atomic finalize_if_running (B-sub-1 review) Follow-up to the /code-review high on the S1 branch. The per-site `if job.status == 'running'` guards were a non-atomic read-then-write (TOCTOU): a cancel committing between the child's status read and its commit could still be overwritten with done/error, re-opening the double-release window. Introduce service.finalize_if_running(db, job_id, **fields): one atomic `UPDATE download_jobs SET ... WHERE status='running'` that returns whether it fired. It flushes pending asset writes first so the asset-then-job row-lock order matches the cancel route (no deadlock), and runs in the caller's session so it commits with the asset writes in one transaction. All five worker sites (download/edit completion, both error paths, cache finish) now go through it, so the invariant lives in one place. The 'job done' logs move inside the finalized branch so a job stopped mid-completion no longer logs a false 'done'. Add tests/test_downloads_finalize.py: running -> finalizes; canceled/paused -> no-op. --- backend/app/downloads/service.py | 31 +++++++++- backend/app/worker.py | 76 ++++++++++-------------- backend/tests/test_downloads_finalize.py | 75 +++++++++++++++++++++++ 3 files changed, 135 insertions(+), 47 deletions(-) create mode 100644 backend/tests/test_downloads_finalize.py diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index ad601cf6..70c6306e 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -8,7 +8,7 @@ the worker live in `worker.py` / the routes; this module owns enqueue + asset bo """ from datetime import datetime, timezone -from sqlalchemy import func, select +from sqlalchemy import func, select, text from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -16,7 +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.downloads.status import HOLDING, JobStatus from app.models import Channel, DownloadJob, MediaAsset, Video @@ -224,6 +224,33 @@ def enqueue_edit( return job +def finalize_if_running(db: Session, job_id: int, **fields) -> bool: + """Write terminal `fields` to a job in ONE atomic ``UPDATE ... WHERE status='running'`` and report + whether it fired. This is the single owner of the invariant "a finishing worker child must never + resurrect a job a route just stopped": the atomic WHERE closes the read-then-write race an + ``if job.status == 'running'`` guard leaves open, so a concurrent cancel/pause can no longer be + overwritten with done/error (which would re-open the double-release → file-delete-under-live-holder + class, since `done`/`error` hold the asset but `canceled` does not — see release_asset). + + Runs in the CALLER's session and does NOT commit — the caller commits it together with its asset + writes in one transaction. It flushes first so any pending asset write takes its row lock BEFORE + this takes the job's, matching the cancel route's asset-then-job order (no deadlock). `fields` keys + are internal column names (never user input). Returns True when the row was still running and got + updated, False when a route had already moved it out of 'running'.""" + if not fields: + return False + db.flush() # asset-then-job lock order: push pending asset writes before we lock the job row + # StrEnum members carry through psycopg as strings, but coerce explicitly so a bound status value + # is always the plain literal the column stores. + params = {k: (v.value if isinstance(v, JobStatus) else v) for k, v in fields.items()} + assignments = ", ".join(f"{col} = :{col}" for col in fields) + result = db.execute( + text(f"UPDATE download_jobs SET {assignments} WHERE id = :_job_id AND status = :_running"), + {**params, "_job_id": job_id, "_running": JobStatus.RUNNING.value}, + ) + return result.rowcount > 0 + + def release_asset(db: Session, job: DownloadJob) -> None: """Drop this job's hold on its asset when it leaves a holding state. Once NO job holds the asset anymore, delete the file + row immediately — a deleted download should free its disk, diff --git a/backend/app/worker.py b/backend/app/worker.py index d7624203..349560ad 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -482,17 +482,15 @@ 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) - # 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 job and asset: + fields = {"status": JobStatus.DONE, "progress": 100, "phase": None} if not job.display_name: # display_name is VARCHAR(255); a title can be longer (e.g. a recipe in the title). - job.display_name = asset.title[:255] if asset.title else None + fields["display_name"] = asset.title[:255] if asset.title else None asset.last_access_at = datetime.now(timezone.utc) + # Atomic finalize: only marks the job done while it is still running, so a concurrent + # cancel/pause is never resurrected (see service.finalize_if_running). + service.finalize_if_running(db, job_id, **fields) db.commit() @@ -872,25 +870,20 @@ 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) - # 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 - job.eta_s = None - if not job.display_name: - # display_name is VARCHAR(255) — clip a long title (see _fill_asset_meta_early). - job.display_name = meta.title[:255] if meta.title else None # Capture the owner before commit expires the row (avoids a refresh SELECT just to notify). owner_id = job.user_id if job else None + fields = {"status": JobStatus.DONE, "progress": 100, "phase": None, "speed_bps": None, + "eta_s": None} + if job and not job.display_name and meta.title: + # display_name is VARCHAR(255) — clip a long title (see _fill_asset_meta_early). + fields["display_name"] = meta.title[:255] + # Atomic terminal write: finalizes ONLY while the row is still running, so a child that + # finishes just after a route cancel/pause can't resurrect it (see finalize_if_running). + finalized = service.finalize_if_running(db, job_id, **fields) db.commit() 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): # Paused/canceled mid-download or mid-transcode: free the asset for a later retry; keep the @@ -904,14 +897,12 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe if asset: asset.status = "error" asset.error = msg - job = db.get(DownloadJob, job_id) - # 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 + # Atomic fail: only while still running, so a route-set cancel/pause is never overwritten + # with `error` (error also holds the asset → would re-open the double-release class). + service.finalize_if_running( + db, job_id, status=JobStatus.ERROR, error=msg, + error_code=dlerrors.classify(msg), phase=None, + ) db.commit() log.warning("job %s failed: %s", job_id, msg) finally: @@ -938,14 +929,11 @@ def _fail_edit(job_id: int, asset_id: int, msg: str) -> None: if asset: asset.status = "error" asset.error = msg - job = db.get(DownloadJob, job_id) - # 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 + # Atomic fail: only while still running (see the download error path). + service.finalize_if_running( + db, job_id, status=JobStatus.ERROR, error=msg, + error_code=dlerrors.classify(msg), phase=None, + ) db.commit() log.warning("edit job %s failed: %s", job_id, msg) @@ -1021,19 +1009,17 @@ 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) - # 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 + # Atomic terminal write (see the download path): don't resurrect a route-stopped job. + finalized = service.finalize_if_running( + db, job_id, status=JobStatus.DONE, progress=100, phase=None + ) db.commit() 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: # Paused/canceled mid-encode: free the asset for a later retry (ref_count is the route's job). diff --git a/backend/tests/test_downloads_finalize.py b/backend/tests/test_downloads_finalize.py new file mode 100644 index 00000000..b4910c44 --- /dev/null +++ b/backend/tests/test_downloads_finalize.py @@ -0,0 +1,75 @@ +"""DB-backed regression tests for `service.finalize_if_running` (B-sub-1). + +The worker's terminal writes go through this helper so a child that finishes just after a route +cancel/pause can NEVER flip the job back to done/error — which would re-open the double-release -> +file-delete-under-live-holder class (done/error hold the asset, canceled does not). The guard is an +atomic `UPDATE ... WHERE status='running'`, so these tests assert it fires exactly when the job is +still running and is a no-op otherwise. Uses the real-Postgres `db` fixture; skipped without a DB. +""" +from sqlalchemy import text + +from app.downloads import service +from app.downloads.status import JobStatus +from app.models import DownloadJob, User + + +def _running_job(db) -> DownloadJob: + user = User(email="dl@example.com") + db.add(user) + db.flush() + job = DownloadJob( + user_id=user.id, + source_kind="youtube", + source_ref="x" * 11, + status=JobStatus.RUNNING.value, + queue_pos=1, + profile_snapshot={}, + ) + db.add(job) + db.commit() + return job + + +def _row(db, job_id): + # Read straight from the DB: SessionLocal uses expire_on_commit=False, so the ORM identity-map + # copy is stale after finalize_if_running's raw UPDATE + commit. + return db.execute( + text("SELECT status, progress FROM download_jobs WHERE id = :id"), {"id": job_id} + ).one() + + +def test_finalize_marks_a_running_job_done_and_reports_it(db): + job = _running_job(db) + won = service.finalize_if_running(db, job.id, status=JobStatus.DONE, progress=100, phase=None) + db.commit() + assert won is True + status, progress = _row(db, job.id) + assert status == "done" + assert progress == 100 + + +def test_finalize_is_a_noop_on_a_canceled_job(db): + # The exact B-sub-1 scenario: a route already canceled the job (released its asset once), and a + # finishing child then tries to write `done`. The guard must NOT resurrect it, or a later delete + # would double-release the asset. + job = _running_job(db) + db.execute( + text("UPDATE download_jobs SET status = 'canceled' WHERE id = :id"), {"id": job.id} + ) + db.commit() + won = service.finalize_if_running(db, job.id, status=JobStatus.DONE, progress=100) + db.commit() + assert won is False + assert _row(db, job.id).status == "canceled" + + +def test_finalize_is_a_noop_on_a_paused_job(db): + job = _running_job(db) + db.execute( + text("UPDATE download_jobs SET status = 'paused' WHERE id = :id"), {"id": job.id} + ) + db.commit() + won = service.finalize_if_running(db, job.id, status=JobStatus.ERROR, error="boom", phase=None) + db.commit() + assert won is False + assert _row(db, job.id).status == "paused"