Merge pull request 'web: report a job finished only once everything finishing it is done' (#27) from job-done-last into main
Reviewed-on: #27 Reviewed-by: branderson <brad@branderson.io>
This commit is contained in:
@@ -77,7 +77,7 @@ def _cleanup_uploads(cfg: Config, job_id: str, video_path: Path | None) -> None:
|
|||||||
Covers are left behind: they are small, and a rebuild inherits cover_path, so
|
Covers are left behind: they are small, and a rebuild inherits cover_path, so
|
||||||
removing one would silently strip #COVER from the rebuilt song. A rebuild can
|
removing one would silently strip #COVER from the rebuilt song. A rebuild can
|
||||||
also inherit its parent's upload, so the directory is not always uploads/<job_id>.
|
also inherit its parent's upload, so the directory is not always uploads/<job_id>.
|
||||||
Never raises — the caller has already marked the job done.
|
Never raises, so cleaning up cannot fail a job that otherwise succeeded.
|
||||||
"""
|
"""
|
||||||
uploads = (cfg.work_dir / "uploads").resolve()
|
uploads = (cfg.work_dir / "uploads").resolve()
|
||||||
dirs = {uploads / job_id}
|
dirs = {uploads / job_id}
|
||||||
@@ -193,6 +193,7 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
|
|||||||
if m := STAGE_LINE.match(msg):
|
if m := STAGE_LINE.match(msg):
|
||||||
job.stage = m[1]
|
job.stage = m[1]
|
||||||
|
|
||||||
|
status, error = "done", None
|
||||||
try:
|
try:
|
||||||
force_from = _prepare_rebuild(job, log)
|
force_from = _prepare_rebuild(job, log)
|
||||||
_write_sidecar(job)
|
_write_sidecar(job)
|
||||||
@@ -201,18 +202,15 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
|
|||||||
job.output_dir = str(result.output_dir)
|
job.output_dir = str(result.output_dir)
|
||||||
job.warnings = list(result.warnings)
|
job.warnings = list(result.warnings)
|
||||||
_drop_superseded_output(job, log)
|
_drop_superseded_output(job, log)
|
||||||
with lock:
|
|
||||||
job.status = "done"
|
|
||||||
# uploads survive a failure so the job stays rerunnable; acquire
|
# uploads survive a failure so the job stays rerunnable; acquire
|
||||||
# has already copied the video into the work dir by now.
|
# has already copied the video into the work dir by now.
|
||||||
_cleanup_uploads(cfg, job_id, job.inp.video_path)
|
_cleanup_uploads(cfg, job_id, job.inp.video_path)
|
||||||
except BaseException as e:
|
except BaseException as e:
|
||||||
with lock:
|
status, error = "failed", str(e)
|
||||||
job.status = "failed"
|
# last, and all at once: a client that sees the job finished must
|
||||||
job.error = str(e)
|
# find everything finishing it does already done
|
||||||
finally:
|
with lock:
|
||||||
with lock:
|
job.status, job.error, job.finished_at = status, error, time.time()
|
||||||
job.finished_at = time.time()
|
|
||||||
|
|
||||||
threading.Thread(target=worker, daemon=True).start()
|
threading.Thread(target=worker, daemon=True).start()
|
||||||
|
|
||||||
|
|||||||
@@ -1852,3 +1852,23 @@ def test_rebuild_all_reports_a_selected_job_that_was_since_rebuilt(tmp_path):
|
|||||||
body = client.post("/api/rebuild-all", data={"force_from": "pitch", "ids": [old]}).json()
|
body = client.post("/api/rebuild-all", data={"force_from": "pitch", "ids": [old]}).json()
|
||||||
assert body == {"ids": [], "skipped": [
|
assert body == {"ids": [], "skipped": [
|
||||||
{"id": old, "reason": "no longer the current job for its song"}]}
|
{"id": old, "reason": "no longer the current job for its song"}]}
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_job_is_only_reported_finished_once_it_has_finished(tmp_path, monkeypatch):
|
||||||
|
"""Done must mean done: a slow upload cleanup used to show a 'done' job
|
||||||
|
with its upload still there and no finished_at."""
|
||||||
|
from songtrace import web
|
||||||
|
real_cleanup = web._cleanup_uploads
|
||||||
|
|
||||||
|
def slow_cleanup(*args):
|
||||||
|
time.sleep(0.3)
|
||||||
|
real_cleanup(*args)
|
||||||
|
|
||||||
|
monkeypatch.setattr(web, "_cleanup_uploads", slow_cleanup)
|
||||||
|
client, cfg = make_client(tmp_path, ok_runner)
|
||||||
|
job_id = client.post("/api/jobs", data={"title": "T", "artist": "A", "lyrics": "x"},
|
||||||
|
files={"video": ("v.mp4", io.BytesIO(b"data"), "video/mp4")}
|
||||||
|
).json()["id"]
|
||||||
|
job = wait_status(client, job_id, "done")
|
||||||
|
assert job["finished_at"] is not None
|
||||||
|
assert not (cfg.work_dir / "uploads" / job_id / "v.mp4").exists()
|
||||||
|
|||||||
Reference in New Issue
Block a user