From ea3098975218b6d08c20f2df571f6aad23d1b4fc Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Sun, 30 Aug 2026 04:39:00 +0300 Subject: [PATCH] app: half the queue was never read /queue answers with queue_running AND queue_pending, and only the first was looked at. A job sits in the second from the moment /prompt returns until the worker picks it up -- always briefly, and for as long as it takes whenever something is running ahead of it, which on a shared box is the ordinary case. A reload in that window found nothing, handed Generate back, and let the same job be queued twice: the first copy then ran unnamed, its messages failing the id check, burning GPU time while the UI narrated the second. Every fixture in client.test.js was cleaner than the real payload -- none of them carried a queue_pending key at all, though ComfyUI always sends both. That is the third time a stand-in tidier than the real thing hid what it stood in for, so the fixtures now carry both halves, including the malformed entries a real answer can contain. Also renamed, because the function no longer answers the question its name asked: queuedFor returns which half it found the job in, and a job waiting its turn is adopted as QUEUED. Saying "working" over a run that has not started is the same small lie as saying "cancelled" over one that has not stopped. --- app/boot.js | 4 +-- app/shell/client.js | 34 +++++++++++++----- app/shell/run.js | 8 +++-- app/shell/session.js | 15 ++++---- app/shell/tests/client.test.js | 64 +++++++++++++++++++++++++++------ app/shell/tests/run.test.js | 11 ++++++ app/shell/tests/session.test.js | 33 ++++++++++++----- 7 files changed, 131 insertions(+), 38 deletions(-) diff --git a/app/boot.js b/app/boot.js index a8ad4810..126ab462 100644 --- a/app/boot.js +++ b/app/boot.js @@ -9,7 +9,7 @@ import { mountAll } from "./shell/panels.js"; import { all as allValues } from "./shell/values.js"; import { composer } from "./composer/composer.js"; import { createRun } from "./shell/run.js"; -import { clientId, connect, runningFor, finishedFor } from "./shell/client.js"; +import { clientId, connect, queuedFor, finishedFor } from "./shell/client.js"; import { wire } from "./shell/session.js"; import { check } from "./shell/pipeline.js"; @@ -23,7 +23,7 @@ async function start() { onGenerate: () => session.generate(), onCancel: () => run.cancel(), }); - const session = wire({ run, page, check, id, runningFor, finishedFor }); + const session = wire({ run, page, check, id, queuedFor, finishedFor }); let manifest; try { diff --git a/app/shell/client.js b/app/shell/client.js index 2b1ba7f3..da01480d 100644 --- a/app/shell/client.js +++ b/app/shell/client.js @@ -40,16 +40,25 @@ export function connect(id) { } /** - * The prompt id this browser already has running, if any. + * What this browser already has in the queue: {promptId, running}, or null. * * Asked of ComfyUI's own queue rather than remembered locally: a run this page * queued may have finished, been cancelled, or been dequeued while the page was * gone, and a remembered id would reattach the UI to a run that no longer - * exists. `queue_running` items are - * [number, prompt_id, prompt, extra_data, outputs], and extra_data carries the - * client_id ComfyUI was given when the run was queued. + * exists. + * + * BOTH halves. `/queue` answers with `queue_running` and `queue_pending`, and a + * job sits in the second one from the moment /prompt returns until the worker + * picks it up -- always briefly, and for as long as it takes whenever something + * is running ahead of it. Reading only the running half meant a reload in that + * window found nothing, handed Generate back, and let the user queue the same + * job twice: the first copy then ran unnamed, its messages failing the id check, + * burning time while the UI narrated the second. + * + * Items in both are [number, prompt_id, prompt, extra_data, outputs], and + * extra_data carries the client_id ComfyUI was given when the run was queued. */ -export async function runningFor(id, { fetch: doFetch = globalThis.fetch, base = "" } = {}) { +export async function queuedFor(id, { fetch: doFetch = globalThis.fetch, base = "" } = {}) { let response; try { response = await doFetch(`${base}/queue`); @@ -59,10 +68,17 @@ export async function runningFor(id, { fetch: doFetch = globalThis.fetch, base = if (!response.ok) return null; const body = await response.json().catch(() => ({})); - for (const item of body.queue_running || []) { - if (!Array.isArray(item) || item.length < 4) continue; - const [, promptId, , extra] = item; - if (extra && extra.client_id === id) return promptId || null; + // Running first: if this browser somehow has both, the one under way is the + // one worth showing. Which half it came from is returned as well, because a + // job waiting its turn is not a job in progress, and telling the user it is + // working when nothing is happening yet is a small lie the UI does not need + // to tell. + for (const [half, running] of [[body.queue_running, true], [body.queue_pending, false]]) { + for (const item of half || []) { + if (!Array.isArray(item) || item.length < 4) continue; + const [, promptId, , extra] = item; + if (extra && extra.client_id === id && promptId) return { promptId, running }; + } } return null; } diff --git a/app/shell/run.js b/app/shell/run.js index aeba8711..231a50fa 100644 --- a/app/shell/run.js +++ b/app/shell/run.js @@ -296,10 +296,14 @@ export function createRun({ * previous page load queued. The id comes from ComfyUI's own queue, so this * adopts a real run rather than assuming one. */ - adopt(promptId) { + adopt(promptId, { running = true } = {}) { if (!promptId) return false; listen(); - emit({ phase: RUNNING, images: [], audio: [], error: null, node: null }); + // A job waiting its turn is queued, not running. Saying "working" over a + // run that has not started is the same small lie as saying "cancelled" + // over one that has not stopped. + emit({ phase: running ? RUNNING : QUEUED, images: [], audio: [], + error: null, node: null }); settle(promptId); return true; }, diff --git a/app/shell/session.js b/app/shell/session.js index a5caa3bb..7c8c0728 100644 --- a/app/shell/session.js +++ b/app/shell/session.js @@ -86,7 +86,7 @@ export function createGenerator({ run, transport, check, ready }) { * ordering is the thing under test, so the ordering lives where tests can reach * it. */ -export function wire({ run, page, check, id, runningFor, finishedFor }) { +export function wire({ run, page, check, id, queuedFor, finishedFor }) { // One subscription draws everything a run affects. It lives here rather than // beside it in boot.js because subscribe() delivers the CURRENT state at // once, and doing that after the button was deliberately disabled handed it @@ -107,7 +107,7 @@ export function wire({ run, page, check, id, runningFor, finishedFor }) { // one on the strength of "phase is idle" is starting one on a guess. page.transport.hold("Looking for a run already in progress"); - const ready = reattach(run, id, { runningFor, finishedFor }).then((adopted) => { + const ready = reattach(run, id, { queuedFor, finishedFor }).then((adopted) => { page.transport.release(run.state); return adopted; }); @@ -116,11 +116,14 @@ export function wire({ run, page, check, id, runningFor, finishedFor }) { return { ready, generate }; } -export async function reattach(run, id, { runningFor, finishedFor } = {}) { +export async function reattach(run, id, { queuedFor, finishedFor } = {}) { try { - const running = await runningFor(id); - if (running) { - if (run.state.phase === "idle") { run.adopt(running); return running; } + const queued = await queuedFor(id); + if (queued) { + if (run.state.phase === "idle") { + run.adopt(queued.promptId, { running: queued.running }); + return queued.promptId; + } return null; } diff --git a/app/shell/tests/client.test.js b/app/shell/tests/client.test.js index 6fbdf015..ffdb6044 100644 --- a/app/shell/tests/client.test.js +++ b/app/shell/tests/client.test.js @@ -3,49 +3,58 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { runningFor, finishedFor } from "../client.js"; +import { queuedFor, finishedFor } from "../client.js"; const answering = (body, ok = true) => async () => ({ ok, status: ok ? 200 : 500, json: async () => body, }); test("the run this browser queued is found by its client id", async () => { - const promptId = await runningFor("me", { + const found = await queuedFor("me", { fetch: answering({ queue_running: [ [1, "someone-elses", {}, { client_id: "them" }, []], [2, "mine", {}, { client_id: "me" }, []], ], + queue_pending: [], }), }); - assert.equal(promptId, "mine"); + assert.deepEqual(found, { promptId: "mine", running: true }); }); test("another browser's run is not adopted", async () => { - const promptId = await runningFor("me", { - fetch: answering({ queue_running: [[1, "theirs", {}, { client_id: "them" }, []]] }), + const promptId = await queuedFor("me", { + fetch: answering({ + queue_running: [[1, "theirs", {}, { client_id: "them" }, []]], + queue_pending: [[2, "also-theirs", {}, { client_id: "them" }, []]], + }), }); assert.equal(promptId, null); }); test("nothing running means nothing to reattach to", async () => { - assert.equal(await runningFor("me", { fetch: answering({ queue_running: [] }) }), null); + assert.equal(await queuedFor("me", { + fetch: answering({ queue_running: [], queue_pending: [] }), + }), null); }); test("a queue that cannot be reached is not an error the app has to show", async () => { // The dev server has no queue behind it, which is the ordinary case while // working on the UI. - const promptId = await runningFor("me", { + const promptId = await queuedFor("me", { fetch: async () => { throw new TypeError("Failed to fetch"); }, }); assert.equal(promptId, null); }); test("a malformed queue entry is skipped rather than crashing the load", async () => { - const promptId = await runningFor("me", { - fetch: answering({ queue_running: [null, "nonsense", [1], [2, "mine", {}, { client_id: "me" }, []]] }), + const found = await queuedFor("me", { + fetch: answering({ + queue_running: [null, "nonsense", [1], [2, "mine", {}, { client_id: "me" }, []]], + queue_pending: [undefined, {}, [3, null, {}, { client_id: "me" }, []]], + }), }); - assert.equal(promptId, "mine"); + assert.deepEqual(found, { promptId: "mine", running: true }); }); test("a finished run is recognised as this browser's from its history entry", async () => { @@ -89,3 +98,38 @@ test("no history to ask is not an error", async () => { }); assert.equal(promptId, null); }); + +test("a run still waiting its turn is found too", async () => { + // /queue answers with both halves, and a job sits in the pending one from the + // moment /prompt returns until the worker picks it up -- always briefly, and + // for as long as it takes whenever something is running ahead of it. A reload + // in that window found nothing and let the same job be queued twice. + const promptId = await queuedFor("me", { + fetch: answering({ + queue_running: [[1, "theirs", {}, { client_id: "them" }, []]], + queue_pending: [[2, "mine", {}, { client_id: "me" }, []]], + }), + }); + assert.deepEqual(promptId, { promptId: "mine", running: false }, + "a job waiting its turn was reported as one already under way"); +}); + +test("a run under way is preferred over one still waiting", async () => { + const promptId = await queuedFor("me", { + fetch: answering({ + queue_running: [[1, "under-way", {}, { client_id: "me" }, []]], + queue_pending: [[2, "waiting", {}, { client_id: "me" }, []]], + }), + }); + assert.deepEqual(promptId, { promptId: "under-way", running: true }); +}); + +test("somebody else's pending run is not adopted either", async () => { + const promptId = await queuedFor("me", { + fetch: answering({ + queue_running: [], + queue_pending: [[1, "theirs", {}, { client_id: "them" }, []]], + }), + }); + assert.equal(promptId, null); +}); diff --git a/app/shell/tests/run.test.js b/app/shell/tests/run.test.js index 9d080675..44d394b6 100644 --- a/app/shell/tests/run.test.js +++ b/app/shell/tests/run.test.js @@ -442,3 +442,14 @@ test("a cancel the server says did nothing does not claim it did", async () => { await run.cancel(); assert.equal(run.state.phase, QUEUED, "the run was declared cancelled on the server's word that it was not"); }); + +test("a run adopted while it is still waiting its turn says Queued, not Working", () => { + const { run } = runner(ok({})); + run.adopt("waiting", { running: false }); + assert.equal(run.state.phase, QUEUED); + assert.equal(run.state.promptId, "waiting"); + + // And it becomes running when the server says it did, not before. + run.handle(JSON.parse(message("execution_start", { prompt_id: "waiting" }).data)); + assert.equal(run.state.phase, RUNNING); +}); diff --git a/app/shell/tests/session.test.js b/app/shell/tests/session.test.js index de442a0a..943191ac 100644 --- a/app/shell/tests/session.test.js +++ b/app/shell/tests/session.test.js @@ -176,7 +176,7 @@ test("a failed attempt leaves Generate usable again", async () => { test("a run still in the queue is taken back over", async () => { const run = fakeRun(); const got = await reattach(run, "me", { - runningFor: async () => "still-going", + queuedFor: async () => ({ promptId: "still-going", running: true }), finishedFor: async () => { throw new Error("history should not have been asked"); }, }); assert.equal(got, "still-going"); @@ -187,7 +187,7 @@ test("a run that finished during the load is found in history", async () => { const run = fakeRun(); run.seen = () => ["ended-just-now"]; const got = await reattach(run, "me", { - runningFor: async () => null, + queuedFor: async () => null, finishedFor: async (_id, seen) => (seen.includes("ended-just-now") ? "ended-just-now" : null), }); assert.equal(got, "ended-just-now"); @@ -197,7 +197,7 @@ test("history is not asked about a run this page never saw", async () => { const run = fakeRun(); let asked = false; const got = await reattach(run, "me", { - runningFor: async () => null, + queuedFor: async () => null, finishedFor: async () => { asked = true; return "something-old"; }, }); assert.equal(got, null); @@ -207,7 +207,7 @@ test("history is not asked about a run this page never saw", async () => { test("a page already running its own generation is not reattached to another", async () => { const run = fakeRun("running"); const got = await reattach(run, "me", { - runningFor: async () => "some-other", + queuedFor: async () => ({ promptId: "some-other", running: true }), finishedFor: async () => null, }); assert.equal(got, null); @@ -216,7 +216,7 @@ test("a page already running its own generation is not reattached to another", a test("no queue and no history is silence, not an error", async () => { const run = fakeRun(); const got = await reattach(run, "me", { - runningFor: async () => { throw new TypeError("Failed to fetch"); }, + queuedFor: async () => { throw new TypeError("Failed to fetch"); }, finishedFor: async () => null, }); assert.equal(got, null); @@ -258,7 +258,7 @@ function wired({ running = null, finished = null, phase = "idle" } = {}) { page: { transport: t }, check: plan(), id: "me", - runningFor: async () => { await queueAnswered; return running; }, + queuedFor: async () => { await queueAnswered; return running ? { promptId: running, running: true } : null; }, finishedFor: async () => finished, }); return { t, run, session, releaseQueue }; @@ -319,7 +319,7 @@ test("a queue that cannot be reached still gives the button back", async () => { const run = fakeRun(); const session = wire({ run, page: { transport: t }, check: plan(), id: "me", - runningFor: async () => { throw new TypeError("Failed to fetch"); }, + queuedFor: async () => { throw new TypeError("Failed to fetch"); }, finishedFor: async () => null, }); await session.ready; @@ -337,7 +337,7 @@ test("a state delivered while the page is still looking does not hand Generate b const run = fakeRun(); wire({ run, page: { transport: t }, check: plan(), id: "me", - runningFor: () => new Promise(() => {}), // never answers + queuedFor: () => new Promise(() => {}), // never answers finishedFor: async () => null, }); @@ -358,7 +358,7 @@ test("the result of a run is shown as it arrives", async () => { run, page: { transport: t, viewer: { setSource: (src, kind) => shown.push([src, kind]) } }, check: plan(), id: "me", - runningFor: async () => null, finishedFor: async () => null, + queuedFor: async () => null, finishedFor: async () => null, }); run.state = { ...run.state, phase: "done", images: [{ filename: "a.png", subfolder: "", type: "output" }] }; @@ -369,3 +369,18 @@ test("the result of a run is shown as it arrives", async () => { run.announce(); assert.equal(shown[1][1], "video", "a video result was handed to an "); }); + +test("a run found waiting in the queue is adopted as waiting", async () => { + const t = transport(); + const run = fakeRun(); + const adopted = []; + run.adopt = (id, opts) => { adopted.push([id, opts]); return true; }; + + const session = wire({ + run, page: { transport: t }, check: plan(), id: "me", + queuedFor: async () => ({ promptId: "waiting", running: false }), + finishedFor: async () => null, + }); + await session.ready; + assert.deepEqual(adopted, [["waiting", { running: false }]]); +});