fix(worker): don't revive a cancelled job when requeuing a busy asset (C-3.10)
The asset-busy and claim-fail branches wrote status='queued' unconditionally, so a pause/cancel the route had just written was clobbered back to queued and the job silently revived. New _requeue_if_running does the requeue as a conditional UPDATE (...WHERE status='running'), a no-op once the route has moved the job off 'running'.
This commit is contained in:
+19
-3
@@ -66,6 +66,22 @@ def _job_status(job_id: int) -> str | None:
|
|||||||
).scalar()
|
).scalar()
|
||||||
|
|
||||||
|
|
||||||
|
def _requeue_if_running(job_id: int, phase: str) -> None:
|
||||||
|
"""Put the job back to 'queued' for a later retry, but ONLY while it is still 'running'. The
|
||||||
|
route flips status to paused/cancelled to stop a job; a blind `status='queued'` here would
|
||||||
|
clobber that and silently revive a job the user just cancelled. The conditional WHERE makes the
|
||||||
|
requeue a no-op unless we still own a running job."""
|
||||||
|
with SessionLocal() as db:
|
||||||
|
db.execute(
|
||||||
|
text(
|
||||||
|
"UPDATE download_jobs SET status='queued', phase=:phase, updated_at=now() "
|
||||||
|
"WHERE id=:id AND status='running'"
|
||||||
|
),
|
||||||
|
{"id": job_id, "phase": phase},
|
||||||
|
)
|
||||||
|
db.commit()
|
||||||
|
|
||||||
|
|
||||||
def _parse_upload_date(raw) -> date | None:
|
def _parse_upload_date(raw) -> date | None:
|
||||||
if not raw:
|
if not raw:
|
||||||
return None
|
return None
|
||||||
@@ -179,13 +195,13 @@ def _process_job(job_id: int) -> None:
|
|||||||
return
|
return
|
||||||
if asset_status == "downloading":
|
if asset_status == "downloading":
|
||||||
# Another worker owns this asset; wait and let the job be reclaimed later.
|
# Another worker owns this asset; wait and let the job be reclaimed later.
|
||||||
_set_job(job_id, status="queued", phase="waiting")
|
_requeue_if_running(job_id, "waiting")
|
||||||
time.sleep(2)
|
time.sleep(2)
|
||||||
return
|
return
|
||||||
|
|
||||||
# asset_status == "pending": try to win the download / edit.
|
# asset_status == "pending": try to win the download / edit.
|
||||||
if not _claim_asset(asset_id):
|
if not _claim_asset(asset_id):
|
||||||
_set_job(job_id, status="queued", phase="waiting")
|
_requeue_if_running(job_id, "waiting")
|
||||||
time.sleep(2)
|
time.sleep(2)
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -414,7 +430,7 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe
|
|||||||
)
|
)
|
||||||
|
|
||||||
layout, retention_days = _download_settings()
|
layout, retention_days = _download_settings()
|
||||||
rel = storage.rel_path(meta, ext, layout)
|
rel = storage.rel_path(meta, ext, layout, asset_id=asset_id)
|
||||||
thumb = _find_thumbnail(staging, produced)
|
thumb = _find_thumbnail(staging, produced)
|
||||||
storage.place_file(produced, settings.download_root, rel)
|
storage.place_file(produced, settings.download_root, rel)
|
||||||
nfo_ok = storage.write_sidecars(settings.download_root, rel, meta, thumb)
|
nfo_ok = storage.write_sidecars(settings.download_root, rel, meta, thumb)
|
||||||
|
|||||||
Reference in New Issue
Block a user