From 04e59550de4331eb3cb69991c32f23df39e1d059 Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:36:42 +0200 Subject: [PATCH] Round 3: a stop during the encoder's last seconds still removes the file; a failed stop request is not a stop Encoder error text no longer says 'upscaled' for Interpolate. --- app/ui/features/interpolate/job.test.js | 9 +++++++-- app/ui/shell/renders.js | 4 +++- core/video_pipe.py | 4 +++- .../output/render/tests/test_jobs_routes.py | 19 +++++++++++++++++++ 4 files changed, 32 insertions(+), 4 deletions(-) diff --git a/app/ui/features/interpolate/job.test.js b/app/ui/features/interpolate/job.test.js index 8edfa6a5..95d172be 100644 --- a/app/ui/features/interpolate/job.test.js +++ b/app/ui/features/interpolate/job.test.js @@ -52,7 +52,7 @@ test("Stop: a job being waited for can be stopped, ends as stopped (not failed, const off = onJobs(() => heard.push(jobsRunning())); let polls = 0; const read = async () => (++polls < 2 ? { state: "running" } : { state: "stopped" }); - const waiting = waitJob(read, "j1", async () => { if (polls === 0) await stopJobs(); }, (id) => stopped.push(id)); + const waiting = waitJob(read, "j1", async () => { if (polls === 0) await stopJobs(); }, (id) => (stopped.push(id), { stopped: true })); await assert.rejects(waiting, (err) => err instanceof JobStopped && err.stopped === true && err.message === "Stopped."); assert.deepEqual(stopped, ["j1"]); assert.deepEqual(heard, [1, 0]); // shown while it waits, hidden when it ends @@ -67,6 +67,11 @@ test("Stop: a stopped job that is read once more is still there (no 404 turned i test("Stop: a chunk that fails while the stop is landing ends as stopped, not as a failure (nothing 'always' follows it)", async () => { let polls = 0; const read = async () => (++polls < 2 ? { state: "running" } : { state: "error", detail: "out of memory" }); - await assert.rejects(waitJob(read, "j3", async () => { if (polls === 0) await stopJobs(); }, () => {}), (err) => err.stopped === true); + await assert.rejects(waitJob(read, "j3", async () => { if (polls === 0) await stopJobs(); }, () => ({ stopped: true })), (err) => err.stopped === true); + // a stop request that failed, or found the job already over, does not turn a real failure into a stop + for (const stop of [() => { throw new Error("network"); }, () => ({ stopped: false })]) { + let n = 0; + await assert.rejects(waitJob(async () => (++n < 2 ? { state: "running" } : { state: "error", detail: "out of memory" }), "j5", async () => { if (n === 0) await stopJobs(); }, stop), /out of memory/); + } await assert.rejects(waitJob(async () => ({ state: "error", detail: "boom" }), "j4", async () => {}), /boom/); }); diff --git a/app/ui/shell/renders.js b/app/ui/shell/renders.js index 9782e04d..2ddc2f3d 100644 --- a/app/ui/shell/renders.js +++ b/app/ui/shell/renders.js @@ -11,7 +11,9 @@ export const jobsRunning = () => active.size; /** -> a function that stops listening. */ export const onJobs = (fn) => { listeners.push(fn); return () => { const i = listeners.indexOf(fn); if (i >= 0) listeners.splice(i, 1); }; }; /** The Stop button: ask every job this page is waiting for to stop, each through its own module's route. */ -export const stopJobs = () => Promise.all([...active].map(([id, stop]) => { asked.add(id); return Promise.resolve(stop(id)).catch(() => {}); })); +export const stopJobs = () => Promise.all([...active].map(async ([id, stop]) => { + try { const r = await stop(id); if (r && r.stopped) asked.add(id); } catch { /* the request failed: the job runs on, and a real failure later is still a failure */ } +})); /** What waitJob throws for a job that was stopped: not a failure, so callers say "stopped" and not "failed". */ export class JobStopped extends Error { constructor() { super("Stopped."); this.stopped = true; } } diff --git a/core/video_pipe.py b/core/video_pipe.py index 3eb3d81b..5ab7f386 100644 --- a/core/video_pipe.py +++ b/core/video_pipe.py @@ -125,10 +125,12 @@ def stream_file(src, dst, upscale, interrupted=lambda: None, fps_factor=1): frames += y.shape[0] if enc is None: raise ClipError(f"no frames could be read from {os.path.basename(src)}") + interrupted() # a stop that came during the last read is still a stop enc.stdin.close() err = enc.stderr.read().decode(errors="replace") if enc.wait() != 0: - raise ClipError(f"ffmpeg could not encode the upscaled video: {err[-500:]}") + raise ClipError(f"ffmpeg could not encode the video: {err[-500:]}") + interrupted() # ... and one that came while the encoder flushed: the file is removed below return frames except BaseException: for p in (dec, enc): diff --git a/modules/output/render/tests/test_jobs_routes.py b/modules/output/render/tests/test_jobs_routes.py index 603458c7..9d579d97 100644 --- a/modules/output/render/tests/test_jobs_routes.py +++ b/modules/output/render/tests/test_jobs_routes.py @@ -214,3 +214,22 @@ def test_the_stop_route_answers_whether_it_stopped_anything(comfy, server): def test_a_job_whose_work_never_checks_cannot_be_stopped(): job = jobs.new() # a final render or an export: it runs to its end assert jobs.stop(job) is False + + +@needs_ffmpeg +def test_a_stop_that_lands_after_the_last_chunk_still_removes_the_file(comfy): + from core import video_pipe + src = comfy.output / "a.mp4" + make_clip(src, seconds=1) + counted = [] + video_pipe.stream_file(str(src), str(comfy.output / "count.mp4"), lambda x: x, lambda: counted.append(1)) + dst = comfy.output / "out.mp4" + calls = [] + + def stop_at_the_last_check(): + calls.append(1) + if len(calls) >= len(counted) - 1: # the checks after the loop: before closing, and after the encoder flushed + raise jobs.Stopped() + with pytest.raises(jobs.Stopped): + video_pipe.stream_file(str(src), str(dst), lambda x: x, stop_at_the_last_check) + assert not dst.exists()