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

@@ -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,

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).

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"