web: report a job finished only once everything finishing it is done #27
@@ -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
|
||||
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>.
|
||||
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()
|
||||
dirs = {uploads / job_id}
|
||||
@@ -193,6 +193,7 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
|
||||
if m := STAGE_LINE.match(msg):
|
||||
job.stage = m[1]
|
||||
|
||||
status, error = "done", None
|
||||
try:
|
||||
force_from = _prepare_rebuild(job, log)
|
||||
_write_sidecar(job)
|
||||
@@ -201,18 +202,15 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
|
||||
job.output_dir = str(result.output_dir)
|
||||
job.warnings = list(result.warnings)
|
||||
_drop_superseded_output(job, log)
|
||||
with lock:
|
||||
job.status = "done"
|
||||
# uploads survive a failure so the job stays rerunnable; acquire
|
||||
# has already copied the video into the work dir by now.
|
||||
_cleanup_uploads(cfg, job_id, job.inp.video_path)
|
||||
except BaseException as e:
|
||||
with lock:
|
||||
job.status = "failed"
|
||||
job.error = str(e)
|
||||
finally:
|
||||
with lock:
|
||||
job.finished_at = time.time()
|
||||
status, error = "failed", str(e)
|
||||
# last, and all at once: a client that sees the job finished must
|
||||
# find everything finishing it does already done
|
||||
with lock:
|
||||
job.status, job.error, job.finished_at = status, error, time.time()
|
||||
|
||||
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()
|
||||
assert body == {"ids": [], "skipped": [
|
||||
{"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