Fix DELETE slug-collision race and widen _discover_jobs error handling
- DELETE now checks under the shared lock whether another job with the same slug is queued/running before removing the job entry, so deleting a done parent can no longer rmtree a rebuild's active work/output directories. - _discover_jobs now wraps the iterdir/stat/read_warnings calls in the same try/except as the JSON parse, so one corrupt or torn sidecar file (e.g. from a killed mid-write) no longer crashes the whole app at boot. - Fixed test_delete_tolerates_a_work_dir_that_is_already_gone, which had become vacuous since _write_sidecar now eagerly creates the work dir; it now removes the dir before asserting DELETE tolerates its absence. - Added a regression test covering the DELETE slug-collision race.
This commit is contained in:
+17
-9
@@ -104,17 +104,18 @@ def _discover_jobs(cfg: Config) -> list[Job]:
|
|||||||
if data.get(key) is not None:
|
if data.get(key) is not None:
|
||||||
data[key] = Path(data[key])
|
data[key] = Path(data[key])
|
||||||
inp = SongInput(**data)
|
inp = SongInput(**data)
|
||||||
|
slug = song_slug(inp)
|
||||||
|
out = cfg.output_dir / slug
|
||||||
|
done = out.exists() and any(out.iterdir())
|
||||||
|
mtime = sidecar.stat().st_mtime
|
||||||
|
warnings = read_warnings(sidecar.parent) if done else []
|
||||||
except (OSError, ValueError, TypeError):
|
except (OSError, ValueError, TypeError):
|
||||||
continue
|
continue
|
||||||
slug = song_slug(inp)
|
|
||||||
out = cfg.output_dir / slug
|
|
||||||
done = out.exists() and any(out.iterdir())
|
|
||||||
mtime = sidecar.stat().st_mtime
|
|
||||||
discovered.append(Job(
|
discovered.append(Job(
|
||||||
id=uuid.uuid4().hex, title=inp.title, artist=inp.artist, inp=inp,
|
id=uuid.uuid4().hex, title=inp.title, artist=inp.artist, inp=inp,
|
||||||
status="done" if done else "failed",
|
status="done" if done else "failed",
|
||||||
output_dir=str(out) if done else None,
|
output_dir=str(out) if done else None,
|
||||||
warnings=read_warnings(sidecar.parent) if done else [],
|
warnings=warnings,
|
||||||
error=None if done else "interrupted before completion",
|
error=None if done else "interrupted before completion",
|
||||||
created_at=mtime, finished_at=mtime))
|
created_at=mtime, finished_at=mtime))
|
||||||
discovered.sort(key=lambda j: j.created_at)
|
discovered.sort(key=lambda j: j.created_at)
|
||||||
@@ -358,14 +359,21 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
|
|||||||
|
|
||||||
@app.delete("/api/jobs/{job_id}", status_code=204)
|
@app.delete("/api/jobs/{job_id}", status_code=204)
|
||||||
def delete_job(job_id: str):
|
def delete_job(job_id: str):
|
||||||
job = _finished_job(job_id)
|
|
||||||
slug = song_slug(job.inp)
|
|
||||||
shutil.rmtree(cfg.work_dir / slug, ignore_errors=True)
|
|
||||||
shutil.rmtree(cfg.output_dir / slug, ignore_errors=True)
|
|
||||||
with lock:
|
with lock:
|
||||||
|
job = jobs.get(job_id)
|
||||||
|
if job is None:
|
||||||
|
raise HTTPException(404)
|
||||||
|
if job.status in ("queued", "running"):
|
||||||
|
raise HTTPException(409, "job is still in progress")
|
||||||
|
slug = song_slug(job.inp)
|
||||||
|
if any(other.status in ("queued", "running") and song_slug(other.inp) == slug
|
||||||
|
for other_id, other in jobs.items() if other_id != job_id):
|
||||||
|
raise HTTPException(409, "another job for this song is in progress")
|
||||||
jobs.pop(job_id, None)
|
jobs.pop(job_id, None)
|
||||||
if job_id in order:
|
if job_id in order:
|
||||||
order.remove(job_id)
|
order.remove(job_id)
|
||||||
|
shutil.rmtree(cfg.work_dir / slug, ignore_errors=True)
|
||||||
|
shutil.rmtree(cfg.output_dir / slug, ignore_errors=True)
|
||||||
|
|
||||||
return app
|
return app
|
||||||
|
|
||||||
|
|||||||
@@ -1070,6 +1070,7 @@ def test_delete_removes_work_and_output_and_the_job(tmp_path):
|
|||||||
def test_delete_tolerates_a_work_dir_that_is_already_gone(tmp_path):
|
def test_delete_tolerates_a_work_dir_that_is_already_gone(tmp_path):
|
||||||
client, cfg = make_client(tmp_path, slug_runner)
|
client, cfg = make_client(tmp_path, slug_runner)
|
||||||
job_id = make_done_job(client)
|
job_id = make_done_job(client)
|
||||||
|
shutil.rmtree(cfg.work_dir / "A - T")
|
||||||
resp = client.delete(f"/api/jobs/{job_id}")
|
resp = client.delete(f"/api/jobs/{job_id}")
|
||||||
assert resp.status_code == 204
|
assert resp.status_code == 204
|
||||||
|
|
||||||
@@ -1097,6 +1098,29 @@ def test_delete_of_a_running_job_is_409(tmp_path):
|
|||||||
assert client.get(f"/api/jobs/{parent}").status_code == 200
|
assert client.get(f"/api/jobs/{parent}").status_code == 200
|
||||||
|
|
||||||
|
|
||||||
|
def test_delete_of_a_job_with_an_active_rebuild_is_409(tmp_path):
|
||||||
|
"""A done job's slug must not be deleted out from under a rebuild in flight."""
|
||||||
|
release = threading.Event()
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
def runner(inp, cfg, log=print, force_from=None, stages=None):
|
||||||
|
calls.append(1)
|
||||||
|
if len(calls) > 1:
|
||||||
|
release.wait(5)
|
||||||
|
return PipelineResult(output_dir=cfg.output_dir / "X")
|
||||||
|
|
||||||
|
client, _ = make_client(tmp_path, runner)
|
||||||
|
parent = make_done_job(client)
|
||||||
|
child = client.post(f"/api/jobs/{parent}/rebuild-full").json()["id"]
|
||||||
|
try:
|
||||||
|
wait_status(client, child, "running")
|
||||||
|
assert client.delete(f"/api/jobs/{parent}").status_code == 409
|
||||||
|
finally:
|
||||||
|
release.set()
|
||||||
|
wait_status(client, child, "done")
|
||||||
|
assert client.get(f"/api/jobs/{parent}").status_code == 200
|
||||||
|
|
||||||
|
|
||||||
def test_a_plain_job_never_removes_anything(tmp_path):
|
def test_a_plain_job_never_removes_anything(tmp_path):
|
||||||
client, cfg = make_client(tmp_path, slug_runner)
|
client, cfg = make_client(tmp_path, slug_runner)
|
||||||
make_done_job(client)
|
make_done_job(client)
|
||||||
|
|||||||
Reference in New Issue
Block a user