fix(downloads): E11.5 epic-end review round 2

Pause now re-reads the job under the same asset-then-job row lock as resume,
cancel, delete and redownload. Its unlocked write waited for a concurrent
cancel's lock and then set the released job back to paused, a holding status,
so a later delete released the same hold again and could free a file another
user's job still held (measured: the overlapping-request test ends 'paused'
instead of 'canceled' with the old read). The lock test now calls the route
handlers, so it also pins which helper each route uses.

Comments follow the code: resolve_url's contract names both verdicts and that a
failed page read is raised as itself; the inherited-failure diagnosis asks the
cookie gate as it stands when the job is claimed, so its advice fits the cookie
that job's Retry would use; the undecryptable-cookie warning no longer claims a
download goes ahead, since the worker parent reaches it only to diagnose.
This commit is contained in:
2026-09-14 01:42:57 +02:00
parent e887233210
commit 2a9ddb78ac
6 changed files with 56 additions and 27 deletions
+8 -4
View File
@@ -35,10 +35,14 @@ class SourceAdapter:
# --- extraction (worker._extract_with_fallback / backfill) ---------------------------------
def resolve_url(self, url: str, ydl_opts: dict) -> str:
"""The link to hand yt-dlp for `url`, resolved BEFORE the first attempt. Raises a RuntimeError
carrying one of this module's verdicts when the source itself shows there is no single video to
download there. `ydl_opts` are the download's own options (cookie, timeouts, address family) for
any page this needs to read. Generic: `url`."""
"""The link to hand yt-dlp for `url`, resolved BEFORE the first attempt. `ydl_opts` are the
download's own options (cookie, timeouts, address family) for any page this needs to read.
Raises a RuntimeError carrying one of this module's verdicts when the page read shows the
download cannot go ahead: the content is there but holds no single video
(`POST_WITHOUT_VIDEO_MESSAGE`), or the page is not the content at all — the login page a
refused cookie gets (`POST_NOT_SHOWN_MESSAGE`), never the no-video verdict. A failed page read
is raised as itself, so it is classified like any other download error. Generic: `url`."""
return url
def client_sets(self, authed: bool = False) -> list:
+4 -2
View File
@@ -229,7 +229,9 @@ def usable_cookie(
) -> str | None:
"""The user's cookie text a download of `url` into `asset` goes with, or None — the ONE decision
whether a download is signed in. The worker child writes it out for yt-dlp (`resolve_cookiefile`);
the worker asks it again for a job that inherits a failed asset, so both jobs are told the same.
the worker asks it again for a job that inherits a failed asset. That second ask sees the
credential as it is when the job is claimed: a cookie the user removed or added since the failed
download changes this job's explanation to fit the cookie its Retry would go with.
The double gate that prevents cross-user leakage:
1. a credential must still exist for `url`'s host (host-scoped via `host_key_for_url`), AND
@@ -252,7 +254,7 @@ def usable_cookie(
if not raw:
log.warning(
"download cookie for user %s host %s could not be decrypted (rotated key?) — "
"proceeding without it",
"treated as no cookie",
user_id,
host_key,
)
+4 -2
View File
@@ -326,8 +326,10 @@ def _extract_with_fallback(
sources make a single attempt. Returns the info dict of the successful attempt.
Before the first attempt the adapter may resolve `url` to the link that really holds the video (a
Facebook group post → its video's watch link); when the source shows no single video there, the
adapter raises its own verdict (adapters.POST_*_MESSAGE) instead of letting yt-dlp guess.
Facebook group post → its video's watch link); when the page shows no single video there, or is not
the post at all (a refused cookie's login page), the adapter raises its own verdict
(adapters.POST_*_MESSAGE) instead of letting yt-dlp guess, and a failed page read fails the
download as itself.
`cookiefile` (S6) is the acting user's per-host cookies.txt, already gated by
`credentials.resolve_cookiefile`; passed to every attempt so the fallback stays authenticated.
+6 -5
View File
@@ -174,10 +174,11 @@ def _own_job(db: Session, user: User, job_id: int) -> DownloadJob:
def _own_job_for_update(db: Session, user: User, job_id: int) -> DownloadJob:
"""`_own_job` for a transition that moves its asset's ref count (resume, cancel, delete, redownload).
The job is re-read under a row lock, so two overlapping requests for one job run one after the other
and the second decides on the first's committed result — never both releasing the same hold, which
can free a file another job still holds.
"""`_own_job` for a transition that moves its asset's ref count (resume, cancel, delete, redownload)
or writes the status those decide on (pause). The job is re-read under a row lock, so two
overlapping requests for one job run one after the other and the second decides on the first's
committed result — never both releasing the same hold, and never a pause turning a job a cancel
just released back into a holder; either can free a file another job still holds.
The asset is locked BEFORE the job: the order a finishing worker child takes them in
(`service.finalize_if_running`), so the two can never deadlock."""
@@ -631,7 +632,7 @@ def update_download_meta(
def pause_download(
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
) -> dict:
job = _own_job(db, user, job_id)
job = _own_job_for_update(db, user, job_id)
service.pause(db, job)
return _serialize_job(db, job)
+4 -4
View File
@@ -418,10 +418,10 @@ def _process_job(job_id: int) -> None:
asset_status = asset.status if asset else None
asset_error = asset.error if asset else None
if asset_status == "error":
# A job that meets an already-failed asset gets the SAME explanation the job that failed
# it got — not a bare "Download failed". Only the text is on the asset, so it is
# diagnosed again, for this link, and whether the user's cookie went with the download
# is asked of the same gate the download child used (`credentials.usable_cookie`).
# A job that meets an already-failed asset gets that failure's explanation — not a bare
# "Download failed". Only the text is on the asset, so it is diagnosed again, for this
# link, with whether a cookie goes with it asked of the same gate the download child
# used (`credentials.usable_cookie`) — as it stands now, so the advice fits this job's Retry.
url = service.source_url(job.source_kind, job.source_ref)
cookie = credentials.usable_cookie(
db, job.user_id, url, asset, formats.normalize(job.profile_snapshot)
+30 -10
View File
@@ -4,23 +4,31 @@ The routes that move a job's asset ref count (resume, cancel, delete, redownload
unlocked read: two overlapping requests for one job (a double-click, a second tab, a replay after a
gateway timeout) both saw it holding its asset and both released the hold. The count then reaches 0
while another job still holds the asset, and GC or the ready-file delete removes it under that job.
`_own_job_for_update` makes the second request wait and decide on the first one's committed result.
A pause racing a cancel did the same one step later: its unlocked write turned the released job back
into a holder, which a later delete releases again. `_own_job_for_update` makes the second request
wait and decide on the first one's committed result.
"""
import threading
import pytest
from fastapi import HTTPException
from sqlalchemy import select
from app.db import SessionLocal
from app.downloads import service
from app.models import DownloadJob, MediaAsset, User
from app.routes.downloads import _own_job_for_update
from app.routes.downloads import (
_own_job_for_update,
cancel_download,
delete_download,
pause_download,
)
_REF = "https://example.com/video"
def _asset_held_by_two_jobs(db):
def _asset_held_by_two_jobs(db, status="error"):
mine, theirs = User(email="lock-mine@example.com"), User(email="lock-theirs@example.com")
db.add_all([mine, theirs])
db.flush()
@@ -35,7 +43,7 @@ def _asset_held_by_two_jobs(db):
asset_id=asset.id,
source_kind="url",
source_ref=_REF,
status="error",
status=status if owner is mine else "error",
queue_pos=0,
profile_snapshot={},
)
@@ -46,23 +54,33 @@ def _asset_held_by_two_jobs(db):
return mine, asset, jobs[0]
@pytest.mark.parametrize("transition", [service.cancel, service.delete])
def test_overlapping_requests_for_one_job_release_its_hold_once(db, transition):
user, asset, job = _asset_held_by_two_jobs(db)
@pytest.mark.parametrize(
("status", "transition", "route", "left"),
[
("error", service.cancel, cancel_download, "canceled"),
("error", service.delete, delete_download, None),
("queued", service.cancel, pause_download, "canceled"),
],
)
def test_overlapping_requests_for_one_job_release_its_hold_once(
db, status, transition, route, left
):
user, asset, job = _asset_held_by_two_jobs(db, status)
job_id = job.id # read before the delete case removes the row under the expired instance
first, second = SessionLocal(), SessionLocal()
finished = threading.Event()
def the_other_request():
try:
transition(second, _own_job_for_update(second, user, job.id))
route(job_id, user=user, db=second)
except HTTPException:
pass # a delete after the delete: the job is gone, a 404
pass # a request after the delete: the job is gone, a 404
finally:
second.close()
finished.set()
try:
locked = _own_job_for_update(first, user, job.id)
locked = _own_job_for_update(first, user, job_id)
other = threading.Thread(target=the_other_request)
other.start()
# The second request waits for the first one's locks instead of deciding on the same row.
@@ -74,6 +92,8 @@ def test_overlapping_requests_for_one_job_release_its_hold_once(db, transition):
first.close()
db.expire_all()
mine = select(DownloadJob.status).where(DownloadJob.id == job_id)
assert db.execute(mine).scalar_one_or_none() == left
assert db.get(MediaAsset, asset.id).ref_count == 1 # the other user's job still holds it