From e74a1cd983e019dfbfcfc62b92717d68ed36da4b Mon Sep 17 00:00:00 2001 From: "[._.]/ Adam Eivy" Date: Wed, 9 Sep 2026 02:44:40 +0000 Subject: [PATCH 1/2] refactor: unify media queue lane admission accounting (#6677) --- server/services/mediaJobQueue/index.js | 36 ++++------- server/services/mediaJobQueue/index.test.js | 70 +++++++++++++++++++++ 2 files changed, 83 insertions(+), 23 deletions(-) diff --git a/server/services/mediaJobQueue/index.js b/server/services/mediaJobQueue/index.js index 40e2a20536..c1ac317cc2 100644 --- a/server/services/mediaJobQueue/index.js +++ b/server/services/mediaJobQueue/index.js @@ -705,29 +705,19 @@ async function drainLoop() { await videoHolds.resolveCohorts(candidates); videoHolds.updateQueued(queue); // Single queue scan, promoting work independently into each open lane. - let gpuOpen = !running; - let cloudSlots = codexParallelLimit - cloudRunning.length; - let remoteSlots = REMOTE_MEDIA_PARALLEL_LIMIT - remoteRunning.length; - if ((gpuOpen || cloudSlots > 0 || remoteSlots > 0) && queue.length > 0) { - for (const job of candidates) { - if (job.status !== 'queued' || job.hold) continue; - const lane = jobLane(job); - if (lane === 'remote') { - if (remoteSlots > 0) { - startLaneJob(job, { lane }); - remoteSlots -= 1; - } - } else if (lane === 'cloud') { - if (cloudSlots > 0) { - startLaneJob(job, { lane }); - cloudSlots -= 1; - } - } else if (gpuOpen) { - startLaneJob(job, { lane }); - gpuOpen = false; - } - if (!gpuOpen && cloudSlots <= 0 && remoteSlots <= 0) break; - } + const limits = laneLimits(); + const slots = { + gpu: limits.gpu - Number(Boolean(running)), + cloud: limits.cloud - cloudRunning.length, + remote: limits.remote - remoteRunning.length, + }; + for (const job of candidates) { + if (job.status !== 'queued' || job.hold) continue; + const lane = jobLane(job); + if (slots[lane] <= 0) continue; + startLaneJob(job, { lane }); + slots[lane] -= 1; + if (Object.values(slots).every((remaining) => remaining <= 0)) break; } await sleep(150); } diff --git a/server/services/mediaJobQueue/index.test.js b/server/services/mediaJobQueue/index.test.js index 0842f6f4f8..d3a6b62de0 100644 --- a/server/services/mediaJobQueue/index.test.js +++ b/server/services/mediaJobQueue/index.test.js @@ -1813,6 +1813,76 @@ describe('local video failure holds', () => { expect(JSON.parse(readFileSync(join(tempDataDir, 'media-jobs.json'), 'utf8')).videoHolds).toEqual([]); }); + it('refills mixed lanes in FIFO order while skipping holds and oversubscribed cloud capacity', async () => { + const heldBatch = Array.from({ length: 4 }, () => submit()); + for (const id of heldBatch.slice(0, 3)) { await tick(); await finish(id); } + mediaJobQueue.setCodexParallelLimit(1); + stubs.generateImageCodex.mockResolvedValue({}); + const cloudJob = () => mediaJobQueue.enqueueJob({ + kind: 'image', params: { mode: 'codex', prompt: 'example cloud image' }, + }).jobId; + const remoteJob = () => submit('example-mlx', { remoteMedia: remoteVideoMediaParams() }); + const gpu = [submit('example-other')]; + const cloud = [cloudJob()]; + // Keep all concurrent remote dispatches on the generator stub. + const { REMOTE_MEDIA_MODULES } = await import('./remoteMediaJob.js'); + vi.spyOn(REMOTE_MEDIA_MODULES, 'video').mockResolvedValue({ + generateVideo: stubs.generateVideoRemote, + }); + const remoteLimit = mediaJobQueue.getQueueCapacity().lanes.remote.limit; + const remote = Array.from({ length: remoteLimit }, remoteJob); + await tick(); + cloud.push(cloudJob()); + expect(mediaJobQueue.runJobNow(cloud[1]).ok).toBe(true); + await tick(); + + // Interleave waiting jobs behind all three busy lanes and the held cohort. + for (let i = 0; i < 2; i += 1) { + cloud.push(cloudJob()); + gpu.push(submit('example-other')); + remote.push(remoteJob()); + } + await tick(); + const occupancy = () => { + const { lanes } = mediaJobQueue.getQueueCapacity(); + return ['gpu', 'cloud', 'remote'].map((lane) => lanes[lane].running); + }; + expect(occupancy()).toEqual([1, 2, remoteLimit]); + expect(mediaJobQueue.getJob(cloud[2]).status).toBe('queued'); + + // Returning cloud from negative to zero slots must still refuse admission; + // GPU and remote refill independently on the same worker pass. + imageGenEvents.emit('completed', { generationId: cloud[0] }); + await finish(gpu[0], null); + await finish(remote[0], null); + await tick(); + expect(occupancy()).toEqual([1, 1, remoteLimit]); + expect(mediaJobQueue.getJob(cloud[2]).status).toBe('queued'); + expect(mediaJobQueue.getJob(gpu[1]).status).toBe('running'); + expect(mediaJobQueue.getJob(remote[remoteLimit]).status).toBe('running'); + expect(mediaJobQueue.getJob(gpu[2]).status).toBe('queued'); + expect(mediaJobQueue.getJob(remote[remoteLimit + 1]).status).toBe('queued'); + + imageGenEvents.emit('completed', { generationId: cloud[1] }); + await finish(gpu[1], null); + await finish(remote[remoteLimit], null); + await tick(); + expect(occupancy()).toEqual([1, 1, remoteLimit]); + expect(mediaJobQueue.getJob(cloud[2]).status).toBe('running'); + expect(mediaJobQueue.getJob(cloud[3]).status).toBe('queued'); + + // A live limit change opens the next cloud slot without restarting the worker. + mediaJobQueue.setCodexParallelLimit(2); + await tick(); + expect(occupancy()).toEqual([1, 2, remoteLimit]); + expect(stubs.generateImageCodex.mock.calls.map(([p]) => p.jobId)).toEqual(cloud); + expect(stubs.generateVideoRemote.mock.calls.map(([p]) => p.jobId)).toEqual(remote); + expect(stubs.generateVideo.mock.calls.slice(3).map(([p]) => p.jobId)).toEqual(gpu); + expect(mediaJobQueue.getJob(heldBatch[3])).toMatchObject({ + status: 'queued', hold: { heldJobCount: 1 }, + }); + }); + it('lets other local models, image, audio, training, cloud and remote work pass a hold', async () => { const ids = Array.from({ length: 4 }, () => submit()); for (const id of ids.slice(0, 3)) { await tick(); await finish(id); } From 776b089aa57ed9651b1950970804381bb42b3c01 Mon Sep 17 00:00:00 2001 From: "[._.]/ Adam Eivy" Date: Wed, 9 Sep 2026 02:47:27 +0000 Subject: [PATCH 2/2] test: document and restore concurrent remote adapter stub (#6677) --- server/services/mediaJobQueue/index.test.js | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/server/services/mediaJobQueue/index.test.js b/server/services/mediaJobQueue/index.test.js index d3a6b62de0..2dbc9ea556 100644 --- a/server/services/mediaJobQueue/index.test.js +++ b/server/services/mediaJobQueue/index.test.js @@ -1824,9 +1824,10 @@ describe('local video failure holds', () => { const remoteJob = () => submit('example-mlx', { remoteMedia: remoteVideoMediaParams() }); const gpu = [submit('example-other')]; const cloud = [cloudJob()]; - // Keep all concurrent remote dispatches on the generator stub. + // Concurrent dynamic imports can bypass the module mock in this test runner; + // without this adapter stub, 19 of 20 dispatches reach the real peer lookup. const { REMOTE_MEDIA_MODULES } = await import('./remoteMediaJob.js'); - vi.spyOn(REMOTE_MEDIA_MODULES, 'video').mockResolvedValue({ + const remoteAdapter = vi.spyOn(REMOTE_MEDIA_MODULES, 'video').mockResolvedValue({ generateVideo: stubs.generateVideoRemote, }); const remoteLimit = mediaJobQueue.getQueueCapacity().lanes.remote.limit; @@ -1881,6 +1882,7 @@ describe('local video failure holds', () => { expect(mediaJobQueue.getJob(heldBatch[3])).toMatchObject({ status: 'queued', hold: { heldJobCount: 1 }, }); + remoteAdapter.mockRestore(); }); it('lets other local models, image, audio, training, cloud and remote work pass a hold', async () => {