towsty/aigen
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Fix queue race: keep the running studio slot until settle finishes, then start the next waiting job.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Towsty
2026-09-05 11:47:10 -05:00
co-authored by Cursor
parent 279fabfb7b
commit e1bad72405
1 changed files with 19 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/utils/studioQueue.ts
+19 -22
View File
@@ -259,12 +259,13 @@ function jobAgeMs(job: { startedAt: number }) {
return Date.now() - job.startedAt
}
/** Uploading, or created just now and not yet on Comfy. Paused leftovers do not count. */
/** Uploading, or created and not yet on Comfy (includes wake / model load). */
export function jobIsLocallySubmitting(job: Job) {
if (job.library?.stopAfterCurrent) return false
if (job.status === 'uploading') return true
if (job.status === 'queued' && !job.promptId && jobAgeMs(job) < SUBMIT_WINDOW_MS) return true
if (job.status === 'running' && !job.promptId && jobAgeMs(job) < SUBMIT_WINDOW_MS) return true
// Must cover ensureComfyReady's busy wait (up to ~3 min). The old 45s window
// let the dispatcher start the next queue item while the current one was still waking Comfy.
if ((job.status === 'queued' || job.status === 'running') && !job.promptId && jobAgeMs(job) < QUEUED_GRACE_MS) return true
return false
}
@@ -339,12 +340,11 @@ function liveJobOwnsGpu(job: Job) {
if (job.library?.chainContinuing) return true
if (job.saving) return true
if (job.status === 'uploading') return true
if (job.status === 'running') {
// Without a prompt id past the submit window, this is a zombie — do not block the queue.
if (!job.promptId && jobAgeMs(job) >= SUBMIT_WINDOW_MS) return false
if (job.status === 'running' || job.status === 'queued') {
// Without a prompt id past the wake/submit grace, this is a zombie — do not block forever.
if (!job.promptId && jobAgeMs(job) >= QUEUED_GRACE_MS) return false
return true
}
if (job.status === 'queued') return Boolean(job.promptId) || jobAgeMs(job) < SUBMIT_WINDOW_MS
return false
}
@@ -397,16 +397,11 @@ export async function videoJobsBusy() {
// An empty Comfy queue is not idle: the 3s buffer + last-frame extract between
// shots leaves Comfy empty while the current video still owns the GPU.
if (listJobs().some(liveJobOwnsGpu)) return true
// Studio "running" only blocks when a live job still owns the GPU, or a fresh
// pending file exists. Orphan disk rows used to block the whole queue forever.
if (listOwnersWithStudioQueues().some(owner => listStudioJobs(owner).some((job) => {
if (job.status !== 'running') return false
if (job.liveJobId) {
const live = getJob(job.liveJobId)
if (live && liveJobOwnsGpu(live)) return true
}
return pendingAlive(job)
}))) return true
// Studio row still "running" owns the queue slot until onLiveVideoSettled clears it.
// Do not peek under the live job here — that raced settle and started the next item early.
if (listOwnersWithStudioQueues().some(owner => listStudioJobs(owner).some(job => job.status === 'running'))) {
return true
}
const queue = await fetchLiveQueue()
if (queue) {
if (queue.running > 0) return true
@@ -825,11 +820,9 @@ function pickHeld(jobs: StudioJob[]) {
}
function pickWaiting(jobs: StudioJob[]) {
const waiting = jobs.filter(job => job.status === 'waiting')
if (!waiting.length) return null
if (jobs.some(job => job.pausedByUser === true)) {
return waiting.reduce((latest, job) => job.createdAt >= latest.createdAt ? job : latest)
}
const waiting = jobs
.filter(job => job.status === 'waiting')
.sort((a, b) => a.createdAt - b.createdAt)
return waiting[0] || null
}
@@ -927,6 +920,9 @@ function repairStaleJobs(jobs: StudioJob[]) {
}
if (job.status !== 'running') continue
const live = job.liveJobId ? getJob(job.liveJobId) : undefined
// Live already complete: onLiveVideoSettled owns finalization. Stealing the row here
// started the next queue item while purge/settle was still running → next jobs failed.
if (live && live.status === 'complete') continue
const liveBusy = live && (live.status === 'queued' || live.status === 'uploading' || live.status === 'running')
if (liveBusy || pendingAlive(job)) continue
const liveFailed = live && (live.status === 'error' || live.status === 'cancelled')
@@ -940,6 +936,7 @@ function repairStaleJobs(jobs: StudioJob[]) {
job.updatedAt = Date.now()
continue
}
// No live job and no fresh pending — orphaned running row.
if (job.shotQueueId) {
job.status = 'held'
job.liveJobId = undefined