feat: remove the output folder a rebuild superseded
This commit is contained in:
@@ -114,6 +114,20 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
|
|||||||
f" ({', '.join(changed) if changed else 'nothing'} changed)")
|
f" ({', '.join(changed) if changed else 'nothing'} changed)")
|
||||||
return force_from
|
return force_from
|
||||||
|
|
||||||
|
def _drop_superseded_output(job: Job, log) -> None:
|
||||||
|
if not job.rebuilt_from:
|
||||||
|
return
|
||||||
|
with lock:
|
||||||
|
parent = jobs.get(job.rebuilt_from)
|
||||||
|
if parent is None or parent.status != "done":
|
||||||
|
return
|
||||||
|
old = rebuild_policy.superseded_output(parent.output_dir, job.output_dir,
|
||||||
|
cfg.output_dir)
|
||||||
|
if old is None:
|
||||||
|
return
|
||||||
|
shutil.rmtree(old, ignore_errors=True)
|
||||||
|
log(f"[rebuild] removed superseded {old.name}")
|
||||||
|
|
||||||
def worker():
|
def worker():
|
||||||
while True:
|
while True:
|
||||||
job_id = todo.get()
|
job_id = todo.get()
|
||||||
@@ -132,9 +146,11 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
|
|||||||
force_from = _prepare_rebuild(job, log)
|
force_from = _prepare_rebuild(job, log)
|
||||||
result = runner(job.inp, cfg, log=log, force_from=force_from)
|
result = runner(job.inp, cfg, log=log, force_from=force_from)
|
||||||
with lock:
|
with lock:
|
||||||
job.status = "done"
|
|
||||||
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)
|
||||||
|
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)
|
||||||
|
|||||||
@@ -821,3 +821,60 @@ def test_rebuild_of_a_running_job_is_409(tmp_path):
|
|||||||
def test_rebuild_of_an_unknown_job_is_404(tmp_path):
|
def test_rebuild_of_an_unknown_job_is_404(tmp_path):
|
||||||
client, _ = make_client(tmp_path, ok_runner)
|
client, _ = make_client(tmp_path, ok_runner)
|
||||||
assert client.post("/api/jobs/nope/rebuild", data=rebuild_form()).status_code == 404
|
assert client.post("/api/jobs/nope/rebuild", data=rebuild_form()).status_code == 404
|
||||||
|
|
||||||
|
|
||||||
|
def slug_runner(inp, cfg, log=print, force_from=None, stages=None):
|
||||||
|
out = cfg.output_dir / song_slug(inp)
|
||||||
|
out.mkdir(parents=True, exist_ok=True)
|
||||||
|
(out / "song.txt").write_text(f"#TITLE:{inp.title}\n")
|
||||||
|
return PipelineResult(output_dir=out)
|
||||||
|
|
||||||
|
|
||||||
|
def test_rebuild_removes_the_output_it_replaced(tmp_path):
|
||||||
|
client, cfg = make_client(tmp_path, slug_runner)
|
||||||
|
parent = make_done_job(client)
|
||||||
|
assert (cfg.output_dir / "A - T" / "song.txt").exists()
|
||||||
|
child = client.post(f"/api/jobs/{parent}/rebuild",
|
||||||
|
data=rebuild_form(title="T2")).json()["id"]
|
||||||
|
job = wait_status(client, child, "done")
|
||||||
|
assert (cfg.output_dir / "A - T2" / "song.txt").exists()
|
||||||
|
assert not (cfg.output_dir / "A - T").exists()
|
||||||
|
assert any("superseded" in line for line in job["log"])
|
||||||
|
|
||||||
|
|
||||||
|
def test_rebuild_keeps_the_output_when_the_slug_is_unchanged(tmp_path):
|
||||||
|
client, cfg = make_client(tmp_path, slug_runner)
|
||||||
|
parent = make_done_job(client)
|
||||||
|
child = client.post(f"/api/jobs/{parent}/rebuild",
|
||||||
|
data=rebuild_form(year="1998")).json()["id"]
|
||||||
|
wait_status(client, child, "done")
|
||||||
|
assert (cfg.output_dir / "A - T" / "song.txt").exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_rebuild_of_a_failed_parent_removes_nothing(tmp_path):
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
def flaky(inp, cfg, log=print, force_from=None, stages=None):
|
||||||
|
calls.append(1)
|
||||||
|
if len(calls) == 1:
|
||||||
|
raise PipelineError("boom")
|
||||||
|
return slug_runner(inp, cfg, log=log)
|
||||||
|
|
||||||
|
client, cfg = make_client(tmp_path, flaky)
|
||||||
|
parent = client.post("/api/jobs", data=rebuild_form()).json()["id"]
|
||||||
|
wait_status(client, parent, "failed")
|
||||||
|
(cfg.output_dir / "A - T").mkdir(parents=True)
|
||||||
|
(cfg.output_dir / "A - T" / "keep.txt").write_text("mine")
|
||||||
|
child = client.post(f"/api/jobs/{parent}/rebuild",
|
||||||
|
data=rebuild_form(title="T2")).json()["id"]
|
||||||
|
wait_status(client, child, "done")
|
||||||
|
assert (cfg.output_dir / "A - T" / "keep.txt").exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_plain_job_never_removes_anything(tmp_path):
|
||||||
|
client, cfg = make_client(tmp_path, slug_runner)
|
||||||
|
make_done_job(client)
|
||||||
|
stray = cfg.output_dir / "A - Other"
|
||||||
|
stray.mkdir(parents=True)
|
||||||
|
make_done_job(client, title="T2")
|
||||||
|
assert stray.exists()
|
||||||
|
|||||||
Reference in New Issue
Block a user