From 4ace48661828344e780fd9c5ee001333aa2953fc Mon Sep 17 00:00:00 2001 From: npeter83 Date: Tue, 18 Aug 2026 21:46:52 +0200 Subject: [PATCH] refactor(downloads): service owns the job state transitions (E3 S2, B-sub-8) pause/resume/cancel/delete/redownload each hand-rolled their asset bookkeeping inline in the route (release, failed-asset reset, the 20-line fresh-asset rebuild), duplicating logic the worker/GC also touch. Move each transition into a service.* method (service.pause/resume/cancel/delete/redownload) that owns the mutation + its commit; the routes keep only auth, the HTTP 409 guards, and serialization. Drops the now-dead _set_status route helper and two stale imports. Pure behaviour-preserving code motion (routes -56 lines). Green under the E3 S2 transition net + the full downloads suite (363 pass), ruff clean. --- backend/app/downloads/service.py | 75 ++++++++++++++++++++++++++++++++ backend/app/routes/downloads.py | 62 +++----------------------- 2 files changed, 81 insertions(+), 56 deletions(-) diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index e3916408..e496c6f2 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -306,3 +306,78 @@ def release_asset(db: Session, job: DownloadJob) -> None: if asset.rel_path: storage.delete_asset_files(settings.download_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 ("queued", "running"): + job.status = "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).""" + if job.status in ("paused", "error"): + job.progress = 0 + job.error = None + job.error_code = None + if job.asset_id: + asset = db.get(MediaAsset, job.asset_id) + if asset and asset.status == "error": + asset.status = "pending" + asset.error = None + job.status = "queued" + db.commit() + + +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 job id + current asset 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).""" + spec = formats.normalize(job.profile_snapshot) + sig = formats.format_sig(spec, salt=f"{job.id}:{job.asset_id}") + asset = get_or_create_asset(db, job.source_kind, job.source_ref, sig) + 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 = "queued" + job.progress = 0 + job.phase = None + job.error = 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 + db.commit() diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index cc435535..ac3c0912 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -21,8 +21,7 @@ from app.config import settings from app.db import get_db from app.downloads import edit as editmod from app.downloads import links as linksmod -from app.downloads import formats, quota, service, storage -from app.downloads.status import JobStatus +from app.downloads import quota, service, storage from app.models import ( DownloadJob, DownloadLink, @@ -454,18 +453,12 @@ def update_download_meta( return _serialize_job(db, job) -def _set_status(db: Session, job: DownloadJob, status: str) -> None: - job.status = status - db.commit() - - @router.post("/{job_id}/pause") def pause_download( job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: job = _own_job(db, user, job_id) - if job.status in ("queued", "running"): - _set_status(db, job, "paused") # a running worker aborts cooperatively via the hook + service.pause(db, job) return _serialize_job(db, job) @@ -474,19 +467,7 @@ def resume_download( job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: job = _own_job(db, user, job_id) - if job.status in ("paused", "error"): - job.progress = 0 - job.error = None - job.error_code = None - # 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 a job whose asset is already errored). - if job.asset_id: - asset = db.get(MediaAsset, job.asset_id) - if asset and asset.status == "error": - asset.status = "pending" - asset.error = None - _set_status(db, job, "queued") + service.resume(db, job) return _serialize_job(db, job) @@ -510,33 +491,7 @@ def redownload_download( raise HTTPException(status_code=409, detail="Editor clips can't be re-downloaded.") if job.status in ("queued", "running", "paused"): raise HTTPException(status_code=409, detail="This download is still in progress.") - - # Resolve the fresh private asset FIRST (before touching any file). Salting with the current - # asset id makes it distinct from both the dedup pool and this job's current asset, so the - # insert never collides — no IntegrityError/rollback can strand us after a file was deleted. - spec = formats.normalize(job.profile_snapshot) - sig = formats.format_sig(spec, salt=f"{job.id}:{job.asset_id}") - asset = service.get_or_create_asset(db, job.source_kind, job.source_ref, sig) - service.populate_from_catalog(db, asset) - - # Now drop the old hold — `service.release_asset` deletes the old file + row only if this job was its - # last holder (a co-held asset is left intact for its other holders). - service.release_asset(db, job) - - service.incr_ref_count(db, asset.id) - job.asset_id = asset.id - job.status = "queued" - job.progress = 0 - job.phase = None - job.error = None - job.speed_bps = None - job.eta_s = None - # Re-derive the display name (card title + Content-Disposition filename) from the freshly - # cleaned title — otherwise a name auto-set from an older, un-cleaned title (e.g. still carrying - # a "1.6M views · … |" social prefix) would survive the rebuild. The worker fills it on completion. - job.display_name = None - job.queue_pos = service.next_queue_pos(db) # tail of the queue, not its stale original slot - db.commit() + service.redownload(db, job) return _serialize_job(db, job) @@ -545,10 +500,7 @@ def cancel_download( job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: job = _own_job(db, user, job_id) - if job.status not in (JobStatus.DONE, JobStatus.CANCELED): - service.release_asset(db, job) # release while the job still counts as holding - job.status = JobStatus.CANCELED - db.commit() + service.cancel(db, job) return _serialize_job(db, job) @@ -557,9 +509,7 @@ def delete_download( job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: job = _own_job(db, user, job_id) - service.release_asset(db, job) - db.delete(job) - db.commit() + service.delete(db, job) return {"deleted": job_id}