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 }]]); +});