peter
815083bcc4
- The queue keeps a done job by when it became done (ready_at), not by its last write: renaming, keeping or moving an old file, or the GC's expiry warning, no longer brings it back as just finished. - GC expiry and eviction re-read each file under the asset→job locks the user routes take, and commit one file at a time: a delete, a re-download or a keep made while the GC runs is honoured, never a second release that frees a file another user still holds. - A file records the origin it was placed under (download_jobs.source_key, migration 0105): an edit whose parent expired still goes back to its folder, not to Uploads. - A folder a file is being put in is held (FOR SHARE) until it lands; a delete of it waits and takes that file home too, instead of failing on the foreign key. - The hash route reads at most two files at once (429 past that) and holds no connection meanwhile. - 0104's backfill dates old files by their request date only: no notice prod wrote names its job. - The old DOWNLOAD_ROOT still finds the files; a files root that is not a mounted volume stops the worker and is logged by the API; the volume-copy recipe looks the project-prefixed names up. - Stale comments about the /downloads mount and the Downloads page.
589 lines
27 KiB
Python
589 lines
27 KiB
Python
"""Enqueue + cache resolution for the download center (the per-user layer over MediaAsset).
|
|
|
|
`enqueue` is the single entry point used by the API: it resolves (or creates) the shared
|
|
MediaAsset for the requested source+format, links a per-user DownloadJob to it, and — on a
|
|
cache hit (asset already ready) — marks the job done instantly with no re-download. The worker
|
|
handles everything else. State transitions used by the routes (pause/resume/cancel/delete) and
|
|
the worker live in `worker.py` / the routes; this module owns enqueue + asset bookkeeping.
|
|
"""
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select, update
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.downloads import adapters
|
|
from app.downloads import clip as clipmod
|
|
from app.downloads import credentials
|
|
from app.downloads import edit as editmod
|
|
from app.downloads import folders, formats, mediakind, quota, storage
|
|
from app.downloads.status import HOLDING, JobKind, JobStatus
|
|
from app.models import (
|
|
DISPLAY_TEXT_MAX,
|
|
Channel,
|
|
DownloadJob,
|
|
MediaAsset,
|
|
PlexItem,
|
|
PlexShow,
|
|
Video,
|
|
)
|
|
from app.plex import lookups as plex_lookups
|
|
from app.plex import paths as plex_paths
|
|
|
|
|
|
class EditError(Exception):
|
|
"""Bad edit request (empty spec / source not ready); `reason` is an i18n-mappable key."""
|
|
|
|
def __init__(self, reason: str):
|
|
self.reason = reason
|
|
super().__init__(reason)
|
|
|
|
|
|
def source_url(source_kind: str, source_ref: str) -> str:
|
|
"""The URL yt-dlp should fetch. YouTube stores the 11-char id (the adapter reconstructs the watch
|
|
URL); ad-hoc sources store the URL verbatim."""
|
|
return adapters.resolve_adapter(source_kind, source_ref).source_url(source_ref)
|
|
|
|
|
|
def reference_url(job, asset) -> str | None:
|
|
"""The clean "downloaded from" URL for a job — shared by the Files library and the public
|
|
watch page so both surfaces always agree. None for edit derivatives (a clip's source is the user's
|
|
own earlier download, not a web page) and for a Plex clip (the library is not a public page); a
|
|
YouTube clip links its video at the clip's start; else the canonical page URL yt-dlp recorded; else
|
|
one derived from source_kind+source_ref (covers queued/older rows before the worker fills it)."""
|
|
if job is None or job.job_kind == JobKind.EDIT:
|
|
return None
|
|
if job.job_kind == JobKind.CLIP:
|
|
return clipmod.reference_url(job.edit_spec or {})
|
|
if asset is not None and asset.source_webpage_url:
|
|
return asset.source_webpage_url
|
|
if job.source_kind in ("youtube", "url"):
|
|
return source_url(job.source_kind, job.source_ref)
|
|
return None
|
|
|
|
|
|
def get_or_create_asset(
|
|
db: Session, source_kind: str, source_ref: str, format_sig: str
|
|
) -> MediaAsset:
|
|
"""Fetch the cache row for this (source, format) or create a fresh `pending` one.
|
|
|
|
Races on the unique cache key are resolved by re-selecting after an IntegrityError, so two
|
|
simultaneous enqueues of the same video+format converge on one shared asset."""
|
|
stmt = select(MediaAsset).where(
|
|
MediaAsset.source_kind == source_kind,
|
|
MediaAsset.source_ref == source_ref,
|
|
MediaAsset.format_sig == format_sig,
|
|
)
|
|
asset = db.execute(stmt).scalar_one_or_none()
|
|
if asset is not None:
|
|
# A previously-FAILED shared asset must be re-attempted for the new job about to hold it —
|
|
# reset it to pending (+clear the error) so the worker actually re-downloads instead of
|
|
# short-circuiting the new job with the asset's stale error. (Mirrors resume_download; without
|
|
# this a fresh enqueue of a once-failed (source,format) pair is permanently poisoned, since
|
|
# errored assets carry no expires_at and GC never clears them.)
|
|
if asset.status == "error":
|
|
asset.status = "pending"
|
|
asset.error = None
|
|
return asset
|
|
asset = MediaAsset(
|
|
source_kind=source_kind,
|
|
source_ref=source_ref,
|
|
format_sig=format_sig,
|
|
status="pending",
|
|
ref_count=0,
|
|
)
|
|
try:
|
|
# A SAVEPOINT: losing the race undoes only this insert, never the caller's own pending writes
|
|
# (Retry clears the job's error before it gets here; a whole-session rollback threw that away).
|
|
with db.begin_nested():
|
|
db.add(asset)
|
|
db.flush()
|
|
# silent-ok: the expected outcome of the race this savepoint exists for — two jobs wanting the
|
|
# same asset. Losing it means the other insert won, and re-selecting is the whole handling.
|
|
except IntegrityError:
|
|
asset = db.execute(stmt).scalar_one()
|
|
return asset
|
|
|
|
|
|
def populate_from_catalog(db: Session, asset: MediaAsset) -> bool:
|
|
"""Fill an asset's display metadata (title/uploader/thumbnail/date/duration) from our own
|
|
catalog when the source is a known YouTube video — so a queued/downloading row shows the
|
|
real title + thumbnail immediately instead of the bare id. 0 network cost. Returns True if
|
|
it found a catalog match."""
|
|
if asset.source_kind != "youtube" or asset.title:
|
|
return False
|
|
video = db.get(Video, asset.source_ref)
|
|
if video is None:
|
|
return False
|
|
asset.title = video.title
|
|
asset.thumbnail_url = video.thumbnail_url
|
|
asset.duration_s = video.duration_seconds
|
|
if video.published_at:
|
|
asset.upload_date = video.published_at.date()
|
|
channel = db.get(Channel, video.channel_id)
|
|
if channel is not None:
|
|
asset.uploader = channel.title
|
|
return True
|
|
|
|
|
|
def next_queue_pos(db: Session) -> int:
|
|
return (db.execute(select(func.coalesce(func.max(DownloadJob.queue_pos), 0))).scalar() or 0) + 1
|
|
|
|
|
|
def incr_ref_count(db: Session, asset_id: int) -> None:
|
|
"""Atomically +1 an asset's hold count AT THE DB. A Python read-modify-write
|
|
(`asset.ref_count = asset.ref_count + 1`) loses an update when two enqueues of the same
|
|
(source,format) run in parallel — both read the same value and write the same +1 — leaving the
|
|
count one short, so the first release can free a file another job still holds (B-sub-5)."""
|
|
db.execute(
|
|
update(MediaAsset)
|
|
.where(MediaAsset.id == asset_id)
|
|
.values(ref_count=MediaAsset.ref_count + 1)
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
|
|
|
|
def decr_ref_count(db: Session, asset_id: int) -> int:
|
|
"""Atomically -1 an asset's hold count (clamped at 0) and RETURN the new value. The caller decides
|
|
whether to free the asset from the returned count while still holding the row's write lock, so two
|
|
concurrent releases can't both observe the same pre-decrement value and double-free — Postgres
|
|
serialises the two UPDATEs on the row lock (B-sub-5). synchronize_session=False: the ORM copy is
|
|
intentionally left stale; callers read the returned value, not `asset.ref_count`."""
|
|
return db.execute(
|
|
update(MediaAsset)
|
|
.where(MediaAsset.id == asset_id)
|
|
.values(ref_count=func.greatest(MediaAsset.ref_count - 1, 0))
|
|
.returning(MediaAsset.ref_count)
|
|
.execution_options(synchronize_session=False)
|
|
).scalar_one()
|
|
|
|
|
|
def enqueue(
|
|
db: Session,
|
|
user_id: int,
|
|
source_kind: str,
|
|
source_ref: str,
|
|
spec: dict,
|
|
profile_id: int | None = None,
|
|
display_name: str | None = None,
|
|
) -> DownloadJob:
|
|
"""Queue a download for `user_id`. Returns the DownloadJob (already `done` on a cache hit).
|
|
|
|
Raises quota.QuotaExceeded if the user is at their job-count or footprint cap."""
|
|
quota.check_enqueue(db, user_id)
|
|
spec = formats.normalize(spec)
|
|
# If this user has a stored cookie for the download's host, salt the cache signature with their
|
|
# id so the cookie-authed result is a PRIVATE asset — never dedup-shared with, or cache-served
|
|
# to, another user (the load-bearing privacy fix; the worker attaches the cookie to exactly this
|
|
# salted identity). A normal download (no matching cookie) gets salt=None and shares as before.
|
|
salt = credentials.salt_for_enqueue(db, user_id, source_url(source_kind, source_ref))
|
|
sig = formats.format_sig(spec, salt=salt)
|
|
asset = get_or_create_asset(db, source_kind, source_ref, sig)
|
|
# Show a real title + thumbnail the moment it's queued (feed downloads are catalog videos).
|
|
populate_from_catalog(db, asset)
|
|
|
|
job = DownloadJob(
|
|
user_id=user_id,
|
|
asset_id=asset.id,
|
|
source_kind=source_kind,
|
|
source_ref=source_ref,
|
|
profile_id=profile_id,
|
|
profile_snapshot=spec,
|
|
display_name=display_name
|
|
or (asset.title[:DISPLAY_TEXT_MAX] if asset.status == "ready" and asset.title else None),
|
|
status=JobStatus.QUEUED,
|
|
queue_pos=next_queue_pos(db),
|
|
)
|
|
db.add(job)
|
|
folders.place(db, job)
|
|
incr_ref_count(db, asset.id)
|
|
|
|
# Cache hit: the file already exists — no worker round-trip needed.
|
|
if asset.status == "ready":
|
|
job.status = JobStatus.DONE
|
|
job.progress = 100
|
|
folders.stamp_done(db, job)
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
|
|
db.commit()
|
|
db.refresh(job)
|
|
return job
|
|
|
|
|
|
def clip_title(source_title: str | None, display_name: str | None) -> str:
|
|
# Clipped to DISPLAY_TEXT_MAX, not to a literal: this feeds `DownloadJob.display_name` through
|
|
# `enqueue_edit`, the same column `display_text` guards. A machine-made title, so it is clipped
|
|
# rather than run through `display_text` — blank here means "no title", not None.
|
|
if display_name:
|
|
return display_name[:DISPLAY_TEXT_MAX]
|
|
base = source_title or "Clip"
|
|
return f"{base} (clip)"[:DISPLAY_TEXT_MAX]
|
|
|
|
|
|
def enqueue_edit(
|
|
db: Session,
|
|
user_id: int,
|
|
source_job: DownloadJob,
|
|
edit_spec: dict,
|
|
display_name: str | None = None,
|
|
) -> DownloadJob:
|
|
"""Queue a phase-2 editor derivative (trim/crop) of `source_job` for `user_id`.
|
|
|
|
The derivative is per-user (never shared-cached): its asset uses `source_kind='edit'` with a
|
|
cache key that embeds the user id, so an identical re-edit by the same user is a cache hit but
|
|
it never dedups across users. Reuses the same worker/quota/GC path as a download.
|
|
|
|
Raises quota.QuotaExceeded (holdings cap) or EditError (empty spec / source not ready)."""
|
|
src_asset = db.get(MediaAsset, source_job.asset_id) if source_job.asset_id else None
|
|
if src_asset is None or src_asset.status != "ready" or not src_asset.rel_path:
|
|
raise EditError("source_not_ready")
|
|
# A generic uploaded file (media_kind='file') has no timeline to trim/crop — the editor path is
|
|
# media-only. Reject here so an upload's file never reaches the ffmpeg edit worker.
|
|
if src_asset.media_kind == mediakind.FILE:
|
|
raise EditError("not_media")
|
|
|
|
spec = editmod.normalize_edit_spec(edit_spec)
|
|
if not spec:
|
|
raise EditError("empty_edit")
|
|
|
|
quota.check_enqueue(db, user_id)
|
|
|
|
sig = editmod.edit_sig(spec)
|
|
ref = f"{user_id}:{src_asset.id}"[:512]
|
|
asset = get_or_create_asset(db, "edit", ref, sig)
|
|
if not asset.title:
|
|
asset.title = clip_title(src_asset.title, display_name)
|
|
asset.uploader = src_asset.uploader
|
|
asset.thumbnail_url = src_asset.thumbnail_url
|
|
asset.upload_date = src_asset.upload_date
|
|
asset.duration_s = editmod.clip_duration(spec, src_asset.duration_s)
|
|
# A clip inherits its source's kind (a video edit stays video, an audio trim stays audio);
|
|
# the ffmpeg edit worker doesn't re-probe codecs, so set it here where the source is known.
|
|
asset.media_kind = src_asset.media_kind or mediakind.VIDEO
|
|
|
|
job = DownloadJob(
|
|
user_id=user_id,
|
|
asset_id=asset.id,
|
|
job_kind=JobKind.EDIT,
|
|
source_kind="edit",
|
|
source_ref=ref,
|
|
source_job_id=source_job.id,
|
|
source_asset_id=src_asset.id,
|
|
edit_spec=spec,
|
|
profile_snapshot={},
|
|
display_name=display_name or asset.title[:DISPLAY_TEXT_MAX],
|
|
status=JobStatus.QUEUED,
|
|
queue_pos=next_queue_pos(db),
|
|
)
|
|
db.add(job)
|
|
folders.place(db, job)
|
|
incr_ref_count(db, asset.id)
|
|
|
|
if asset.status == "ready": # identical clip already produced for this user → instant
|
|
job.status = JobStatus.DONE
|
|
job.progress = 100
|
|
folders.stamp_done(db, job)
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
|
|
db.commit()
|
|
db.refresh(job)
|
|
return job
|
|
|
|
|
|
def _clip_source_facts(db: Session, spec: dict) -> dict:
|
|
"""Title, uploader, thumbnail and length of what a clip is cut from, read from our own catalog.
|
|
Raises ClipError when a Plex item is unknown or its file cannot be read here — said at enqueue,
|
|
rather than as a job that fails in the queue a minute later."""
|
|
if spec["source"] == "youtube":
|
|
video = db.get(Video, spec["ref"])
|
|
if video is None: # not in the catalog: titled "Clip" until yt-dlp names it (worker)
|
|
return {"title": None, "uploader": None, "thumbnail_url": None, "length_s": None}
|
|
channel = db.get(Channel, video.channel_id)
|
|
return {
|
|
"title": video.title,
|
|
"uploader": channel.title if channel else None,
|
|
"thumbnail_url": video.thumbnail_url,
|
|
"length_s": video.duration_seconds,
|
|
}
|
|
item = plex_lookups.by_rating_key(db, PlexItem, spec["ref"])
|
|
if item is None:
|
|
raise clipmod.ClipError("not_found")
|
|
if plex_paths.local_media_path(db, item.file_path) is None:
|
|
raise clipmod.ClipError("media_missing")
|
|
title = item.title
|
|
show = db.get(PlexShow, item.show_id) if item.show_id else None
|
|
if show is not None:
|
|
episode = (
|
|
f"S{item.season_number}·E{item.episode_number}"
|
|
if item.season_number is not None and item.episode_number is not None
|
|
else None
|
|
)
|
|
title = " — ".join(p for p in (show.title, episode, item.title) if p)
|
|
return {
|
|
"title": title,
|
|
"uploader": None,
|
|
# Plex art is served through our own authenticated routes, never a public URL; the worker
|
|
# cuts the clip's poster from the clip itself.
|
|
"thumbnail_url": None,
|
|
"length_s": item.duration_s,
|
|
}
|
|
|
|
|
|
def enqueue_clip(
|
|
db: Session, user_id: int, spec: dict, display_name: str | None = None
|
|
) -> DownloadJob:
|
|
"""Queue an ad-hoc clip (`clipmod.normalize`d spec) for `user_id`: a NEW per-user asset cut
|
|
straight from the source — a YouTube range or a Plex library file (see downloads/clip).
|
|
|
|
Raises ClipError (unknown or unreachable source, a range past its end) or quota.QuotaExceeded."""
|
|
facts = _clip_source_facts(db, spec)
|
|
# One second of grace: a mark set on the last frame may round past the catalog's whole seconds.
|
|
if facts["length_s"] and spec["end_s"] > facts["length_s"] + 1:
|
|
raise clipmod.ClipError("past_end")
|
|
|
|
quota.check_enqueue(db, user_id)
|
|
|
|
ref = clipmod.asset_ref(user_id, spec)[:512]
|
|
asset = get_or_create_asset(db, "clip", ref, clipmod.sig(spec))
|
|
if not asset.title:
|
|
asset.title = clip_title(facts["title"], display_name)
|
|
asset.uploader = facts["uploader"]
|
|
asset.thumbnail_url = facts["thumbnail_url"]
|
|
asset.duration_s = round(clipmod.duration(spec))
|
|
asset.media_kind = mediakind.VIDEO
|
|
|
|
job = DownloadJob(
|
|
user_id=user_id,
|
|
asset_id=asset.id,
|
|
job_kind=JobKind.CLIP,
|
|
source_kind="clip",
|
|
source_ref=ref,
|
|
edit_spec=spec,
|
|
profile_snapshot={},
|
|
display_name=display_name or asset.title[:DISPLAY_TEXT_MAX],
|
|
status=JobStatus.QUEUED,
|
|
queue_pos=next_queue_pos(db),
|
|
)
|
|
db.add(job)
|
|
folders.place(db, job)
|
|
incr_ref_count(db, asset.id)
|
|
|
|
if asset.status == "ready": # the same range was already cut for this user → instant
|
|
job.status = JobStatus.DONE
|
|
job.progress = 100
|
|
folders.stamp_done(db, job)
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
|
|
db.commit()
|
|
db.refresh(job)
|
|
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` are
|
|
mapped-column kwargs (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
|
|
if fields.get("status") == JobStatus.DONE:
|
|
fields = {**fields, "ready_at": datetime.now(timezone.utc)}
|
|
# Done is when a file's retention starts: its folder's, unless the user kept it.
|
|
held = db.execute(
|
|
select(DownloadJob.folder_id, DownloadJob.kept).where(DownloadJob.id == job_id)
|
|
).one_or_none()
|
|
if held is not None:
|
|
fields = {
|
|
**fields,
|
|
"expires_at": folders.done_expiry(db, held.folder_id, held.kept),
|
|
"expiry_warned": False,
|
|
}
|
|
# A Core UPDATE with the guard as a WHERE — same atomic "only while still running" write as the
|
|
# rest of the codebase's 2.0 statements, no hand-built SQL string. synchronize_session=False: this
|
|
# is a terminal write, so we don't pay to sync the (about-to-be-discarded) ORM identity map. Bound
|
|
# StrEnum values bind as their plain str literal (StrEnum is a str subclass).
|
|
result = db.execute(
|
|
update(DownloadJob)
|
|
.where(DownloadJob.id == job_id, DownloadJob.status == JobStatus.RUNNING.value)
|
|
.values(**fields)
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
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,
|
|
and the shared cache only needs to span *overlapping* holders (a later re-add just downloads
|
|
again). `error` counts as holding: enqueue always +1'd ref_count and the worker never
|
|
decrements on failure (ref_count is the route's job), so an errored job releases here on
|
|
delete — else its +1 leaks and a later resume→ready keeps the file pinned past its last holder.
|
|
Idempotent for the cancel path: cancel already released while the job was holding, then set it
|
|
`canceled` (∉ the set below), so delete-after-cancel is a no-op and never double-releases.
|
|
|
|
We DON'T delete an errored asset row here even at ref==0: another request may be concurrently
|
|
re-enqueuing the same (source,format) — get_or_create_asset would reset that row to `pending` and
|
|
+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.
|
|
|
|
An UPLOAD is the exception: it is never shared or re-enqueued (its ref is unique), and while it
|
|
waits for or failed its phone-safe conversion the original file already sits at `rel_path` — so
|
|
it is freed at ref==0 whatever its state, except mid-conversion (`downloading`: the worker child
|
|
still reads it; the GC upload pass takes it once the child has stopped)."""
|
|
if not job.asset_id or job.status not in HOLDING:
|
|
return
|
|
asset = db.get(MediaAsset, job.asset_id)
|
|
if asset is None:
|
|
return
|
|
new_count = decr_ref_count(db, asset.id)
|
|
freeable = asset.status == "ready" or (
|
|
asset.source_kind == "upload" and asset.status != "downloading"
|
|
)
|
|
if new_count == 0 and freeable:
|
|
if asset.rel_path:
|
|
storage.delete_asset_files(settings.files_root, asset.rel_path)
|
|
db.delete(asset)
|
|
|
|
|
|
# --- Job state transitions ---------------------------------------------------------------------
|
|
# The single home for the pause/resume/cancel/delete/redownload transitions + their asset
|
|
# bookkeeping (release, failed-asset reset, fresh-asset rebuild), so the routes are thin callers and
|
|
# the worker/GC could share the same definitions (B-sub-8). Each commits its own transition; the
|
|
# route keeps only auth + HTTP mapping + serialization.
|
|
|
|
|
|
def pause(db: Session, job: DownloadJob) -> None:
|
|
"""Pause a queued/running job (a running worker aborts cooperatively via the cancel poll); a no-op
|
|
in any other state."""
|
|
if job.status in (JobStatus.QUEUED, JobStatus.RUNNING):
|
|
job.status = JobStatus.PAUSED
|
|
db.commit()
|
|
|
|
|
|
def resume(db: Session, job: DownloadJob) -> None:
|
|
"""Requeue a paused/errored job for a fresh attempt, clearing its error. If the SHARED asset itself
|
|
failed, reset it to pending so the worker actually re-downloads — otherwise the requeued job would
|
|
instantly re-inherit the asset's stale error (the worker short-circuits an errored asset).
|
|
|
|
A cookie the user added AFTER the job was queued moves the job onto their private asset first
|
|
(`_follow_the_cookie`): the worker only ever attaches a cookie to that identity, so without the move
|
|
the Retry a login-wall message asks for would fail on the same wall."""
|
|
if job.status in (JobStatus.PAUSED, JobStatus.ERROR):
|
|
job.progress = 0
|
|
job.error = None
|
|
job.error_code = None
|
|
job.error_detail = None
|
|
if job.asset_id:
|
|
asset = db.get(MediaAsset, job.asset_id)
|
|
if asset is not None:
|
|
asset = _follow_the_cookie(db, job, asset)
|
|
if asset and asset.status == "error": # MediaAsset.status (no enum), not a JobStatus
|
|
asset.status = "pending"
|
|
asset.error = None
|
|
job.status = JobStatus.QUEUED
|
|
db.commit()
|
|
|
|
|
|
def _follow_the_cookie(db: Session, job: DownloadJob, asset: MediaAsset) -> MediaAsset:
|
|
"""Re-point a requeued job at the asset enqueue would pick NOW, when the user has a cookie for the
|
|
link's host but the job still holds an asset the worker won't attach it to (queued before the
|
|
cookie existed). Returns the asset the job holds afterwards.
|
|
|
|
Only a `pending` / `error` asset moves — neither has a file nor a download in flight. A `ready` one
|
|
already has the file, and a `downloading` one may be this job's own child still being stopped by
|
|
the pause; the next Retry after it settles moves the job then. Removing a cookie never moves a job
|
|
back: the worker attaches nothing without a credential, so the private asset is as good as any."""
|
|
if asset.status not in ("pending", "error"):
|
|
return asset
|
|
url = source_url(job.source_kind, job.source_ref)
|
|
salt = credentials.salt_for_enqueue(db, job.user_id, url)
|
|
spec = formats.normalize(job.profile_snapshot)
|
|
if salt is None or credentials.is_authed_asset(asset, spec, job.user_id, url):
|
|
return asset
|
|
target = get_or_create_asset(
|
|
db, job.source_kind, job.source_ref, formats.format_sig(spec, salt=salt)
|
|
)
|
|
populate_from_catalog(db, target)
|
|
release_asset(db, job) # drop the old hold BEFORE re-pointing — it keys off job.asset_id
|
|
incr_ref_count(db, target.id)
|
|
job.asset_id = target.id
|
|
return target
|
|
|
|
|
|
def cancel(db: Session, job: DownloadJob) -> None:
|
|
"""Cancel a job unless it's already done/canceled, releasing its asset hold WHILE it still counts
|
|
as holding — `canceled` ∉ HOLDING, so a later delete won't double-release."""
|
|
if job.status not in (JobStatus.DONE, JobStatus.CANCELED):
|
|
release_asset(db, job) # release while the job still counts as holding
|
|
job.status = JobStatus.CANCELED
|
|
db.commit()
|
|
|
|
|
|
def delete(db: Session, job: DownloadJob) -> None:
|
|
"""Delete a job, releasing its asset hold first (which frees the file iff this was the last holder)."""
|
|
release_asset(db, job)
|
|
db.delete(job)
|
|
db.commit()
|
|
|
|
|
|
def redownload(db: Session, job: DownloadJob) -> None:
|
|
"""Rebuild a finished/failed job's file into a FRESH, private asset (the cache sig is salted with
|
|
the new asset's own id, so it never reuses the dedup pool) and requeue the SAME job row —
|
|
so any public share link (DownloadLink -> job_id) keeps resolving to the newly-downloaded file.
|
|
The old hold is dropped first (freeing the old file iff this job was its last holder). The caller
|
|
has already rejected edit clips + still-in-progress jobs (HTTP 409).
|
|
|
|
When the user has a cookie for the link's host, the salt carries it (`credentials.rebuild_salt`),
|
|
so the worker recognises the rebuild as their private identity and attaches the cookie — a cookie
|
|
download, or one the user added after a failure, rebuilds with the cookie."""
|
|
spec = formats.normalize(job.profile_snapshot)
|
|
cookie = credentials.salt_for_enqueue(
|
|
db, job.user_id, source_url(job.source_kind, job.source_ref)
|
|
)
|
|
# The salt names the row's own id, which exists only once the row does: insert it under a unique
|
|
# placeholder, then write the real signature in the same transaction (nobody sees the placeholder).
|
|
asset = MediaAsset(
|
|
source_kind=job.source_kind,
|
|
source_ref=job.source_ref,
|
|
format_sig=f"rebuild:{uuid.uuid4().hex}",
|
|
status="pending",
|
|
ref_count=0,
|
|
)
|
|
db.add(asset)
|
|
db.flush()
|
|
asset.format_sig = formats.format_sig(spec, salt=credentials.rebuild_salt(asset.id, cookie))
|
|
db.flush()
|
|
populate_from_catalog(db, asset)
|
|
# Drop the old hold BEFORE re-pointing the job — release_asset keys off the job's current asset.
|
|
release_asset(db, job)
|
|
incr_ref_count(db, asset.id)
|
|
job.asset_id = asset.id
|
|
job.status = JobStatus.QUEUED
|
|
job.progress = 0
|
|
job.phase = None
|
|
job.error = None
|
|
job.error_code = None # a rebuild that fails again gets its own diagnosis, never the old one
|
|
job.error_detail = None
|
|
job.speed_bps = None
|
|
job.eta_s = None
|
|
# Re-derive the display name from the freshly cleaned title on completion (else an older, un-cleaned
|
|
# name would survive the rebuild).
|
|
job.display_name = None
|
|
job.queue_pos = next_queue_pos(db) # tail of the queue, not its stale original slot
|
|
job.expires_at = None # stamped again when the rebuild is done
|
|
job.expiry_warned = False
|
|
db.commit()
|