From 9e6db98195414b5b9c74c8d8ab65648f4d4109de Mon Sep 17 00:00:00 2001 From: 134 Date: Mon, 3 Aug 2026 13:47:56 +0800 Subject: [PATCH] fix(queue): improve image throughput and live status --- app/(chat)/api/chat/_lib/stream.ts | 6 ++ app/(chat)/api/chat/route.ts | 8 ++ components/canvas/CanvasChatPanel.tsx | 10 ++ .../canvas/chat-panel/canvas-agent-stream.ts | 1 + components/image-generation-loading.tsx | 21 +++- lib/canvas/image-concurrency.ts | 99 ++++++++++++++----- lib/canvas/image-queue-fairness.ts | 11 +++ lib/canvas/image-task-queue.ts | 54 +++++++--- playwright.helpers.config.ts | 1 + .../canvas-image-queue-fairness.test.ts | 37 +++++++ 10 files changed, 208 insertions(+), 40 deletions(-) create mode 100644 lib/canvas/image-queue-fairness.ts create mode 100644 tests/routes/canvas-image-queue-fairness.test.ts diff --git a/app/(chat)/api/chat/_lib/stream.ts b/app/(chat)/api/chat/_lib/stream.ts index 66ba6746d..2b999ad7a 100644 --- a/app/(chat)/api/chat/_lib/stream.ts +++ b/app/(chat)/api/chat/_lib/stream.ts @@ -211,6 +211,7 @@ export function writeImageGenerationStatusChunk( modelName?: string | null; imageCount?: number | null; queuePosition?: number | null; + sameModelPendingCount?: number | null; sameModelRunningCount?: number | null; concurrencyLimit?: number | null; estimatedWaitMs?: number | null; @@ -233,6 +234,11 @@ export function writeImageGenerationStatusChunk( Number(params.queuePosition) >= 0 ? Number(params.queuePosition) : undefined, + sameModelPendingCount: + Number.isFinite(Number(params.sameModelPendingCount)) && + Number(params.sameModelPendingCount) >= 0 + ? Number(params.sameModelPendingCount) + : undefined, sameModelRunningCount: Number.isFinite(Number(params.sameModelRunningCount)) && Number(params.sameModelRunningCount) >= 0 diff --git a/app/(chat)/api/chat/route.ts b/app/(chat)/api/chat/route.ts index 14b0ecb61..26231236f 100644 --- a/app/(chat)/api/chat/route.ts +++ b/app/(chat)/api/chat/route.ts @@ -6035,6 +6035,8 @@ export async function POST(request: Request) { resolvedModelRef, imageCount: 1, queuePosition: queueStatus.queuePosition, + sameModelPendingCount: + queueStatus.sameModelPendingCount, sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, @@ -6934,6 +6936,8 @@ export async function POST(request: Request) { resolvedModelRef, imageCount: 1, queuePosition: queueStatus.queuePosition, + sameModelPendingCount: + queueStatus.sameModelPendingCount, sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, estimatedWaitMs: queueStatus.estimatedWaitMs, @@ -7259,6 +7263,8 @@ export async function POST(request: Request) { queueStatus.modelName || selectedAgentPreset.name, imageCount: resolvedImageCount, queuePosition: queueStatus.queuePosition, + sameModelPendingCount: + queueStatus.sameModelPendingCount, sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, estimatedWaitMs: queueStatus.estimatedWaitMs, @@ -7486,6 +7492,8 @@ export async function POST(request: Request) { modelName: queueStatus.modelName || slotLabel, imageCount: 1, queuePosition: queueStatus.queuePosition, + sameModelPendingCount: + queueStatus.sameModelPendingCount, sameModelRunningCount: queueStatus.sameModelRunningCount, concurrencyLimit: queueStatus.concurrencyLimit, estimatedWaitMs: queueStatus.estimatedWaitMs, diff --git a/components/canvas/CanvasChatPanel.tsx b/components/canvas/CanvasChatPanel.tsx index fc893dce9..01dff07e9 100644 --- a/components/canvas/CanvasChatPanel.tsx +++ b/components/canvas/CanvasChatPanel.tsx @@ -7175,6 +7175,11 @@ export function CanvasChatPanel({ Number(payload.queuePosition) >= 0 ? Number(payload.queuePosition) : undefined, + sameModelPendingCount: + Number.isFinite(Number(payload.sameModelPendingCount)) && + Number(payload.sameModelPendingCount) >= 0 + ? Number(payload.sameModelPendingCount) + : undefined, sameModelRunningCount: Number.isFinite(Number(payload.sameModelRunningCount)) && Number(payload.sameModelRunningCount) >= 0 @@ -13689,6 +13694,8 @@ export function CanvasChatPanel({ ? streamImageGenerationState.phase : "processing"; const loadingQueuePosition = streamImageGenerationState?.queuePosition; + const loadingSameModelPendingCount = + streamImageGenerationState?.sameModelPendingCount; const loadingSameModelRunningCount = streamImageGenerationState?.sameModelRunningCount; const loadingConcurrencyLimit = streamImageGenerationState?.concurrencyLimit; @@ -13704,6 +13711,7 @@ export function CanvasChatPanel({ modelName: loadingModelName, imageCount: loadingImageCount, queuePosition: loadingQueuePosition, + sameModelPendingCount: loadingSameModelPendingCount, sameModelRunningCount: loadingSameModelRunningCount, concurrencyLimit: loadingConcurrencyLimit, estimatedWaitMs: loadingEstimatedWaitMs, @@ -13716,6 +13724,7 @@ export function CanvasChatPanel({ loadingPhase, loadingProviderType, loadingQueuePosition, + loadingSameModelPendingCount, loadingSameModelRunningCount, shouldShowImageLoading, streamImageGenerationState, @@ -15439,6 +15448,7 @@ export function CanvasChatPanel({ modelName={loadingModelName} phase={loadingPhase} queuePosition={loadingQueuePosition} + sameModelPendingCount={loadingSameModelPendingCount} sameModelRunningCount={loadingSameModelRunningCount} concurrencyLimit={loadingConcurrencyLimit} estimatedWaitMs={loadingEstimatedWaitMs} diff --git a/components/canvas/chat-panel/canvas-agent-stream.ts b/components/canvas/chat-panel/canvas-agent-stream.ts index e7a6246a2..192765f7a 100644 --- a/components/canvas/chat-panel/canvas-agent-stream.ts +++ b/components/canvas/chat-panel/canvas-agent-stream.ts @@ -155,6 +155,7 @@ export type ImageGenerationStatusStreamPayload = { modelName?: string; imageCount?: number; queuePosition?: number; + sameModelPendingCount?: number; sameModelRunningCount?: number; concurrencyLimit?: number; estimatedWaitMs?: number; diff --git a/components/image-generation-loading.tsx b/components/image-generation-loading.tsx index 20ce806dc..52b597454 100644 --- a/components/image-generation-loading.tsx +++ b/components/image-generation-loading.tsx @@ -15,6 +15,7 @@ interface ImageGenerationLoadingProps { phase?: "idle" | "queued" | "processing" | "finalizing" | "failed"; elapsedSeconds?: number; queuePosition?: number; + sameModelPendingCount?: number; sameModelRunningCount?: number; concurrencyLimit?: number; estimatedWaitMs?: number; @@ -68,6 +69,7 @@ const LONG_PROCESSING_HINTS = [ "正在准备把结果显示到画布。", ]; const LONG_WAIT_SECONDS = 5 * 60; +const LONG_QUEUE_WAIT_SECONDS = 10 * 60; const LONG_WAIT_HINT = "服务繁忙，模型可能正在排队或自动尝试备用通道，请耐心等候，结果返回后会自动同步到画布。"; @@ -106,6 +108,7 @@ export const ImageGenerationLoading = memo( phase = "processing", elapsedSeconds, queuePosition, + sameModelPendingCount, sameModelRunningCount, concurrencyLimit, estimatedWaitMs, @@ -138,6 +141,8 @@ export const ImageGenerationLoading = memo( const isLongWait = resolvedElapsedSeconds >= LONG_WAIT_SECONDS && (phase === "processing" || phase === "finalizing"); + const isLongQueueWait = + phase === "queued" && resolvedElapsedSeconds >= LONG_QUEUE_WAIT_SECONDS; const estimateRangeLabel = isLongWait ? "单次最长约10分钟" : providerType === "grsai-gpt-image" @@ -156,13 +161,22 @@ export const ImageGenerationLoading = memo( : null; const queueDetailLine = phase === "queued" - ? `当前模型运行 ${Math.max( + ? `队列共 ${Math.max( + 0, + Number(sameModelPendingCount || 0) + )} 个待处理，当前模型运行 ${Math.max( 0, Number(sameModelRunningCount || 0) )}/${Math.max(1, Number(concurrencyLimit || 0) || 1)}，${formatImageQueueWaitLabel( - estimatedWaitMs + Math.max( + 0, + Number(estimatedWaitMs || 0) - resolvedElapsedSeconds * 1000 + ) )}` : null; + const queueTimeoutHint = isLongQueueWait + ? "排队时间较长，队列位置仍会实时更新；进入执行后单次最长约10分钟，超时会自动提示。" + : null; return (
{queueLine ?
{queueLine}
: null} {queueDetailLine ?
{queueDetailLine}
: null} + {queueTimeoutHint ? ( +
{queueTimeoutHint}
+ ) : null} {(phase !== "processing" || resolvedElapsedSeconds > 0) && (
{phaseHint} {elapsedLabel} diff --git a/lib/canvas/image-concurrency.ts b/lib/canvas/image-concurrency.ts index 1fe03a218..3233adfe2 100644 --- a/lib/canvas/image-concurrency.ts +++ b/lib/canvas/image-concurrency.ts @@ -8,6 +8,7 @@ import { type CanvasImageTaskQueueConfig, normalizeCanvasImageTaskQueueConfig, } from "./image-task-config"; +import { getAdaptiveImageQueueOwnerLimit } from "./image-queue-fairness"; const CONCURRENCY_CONFIG_CACHE_MS = 5 * 1000; const RECENT_EXECUTION_SAMPLE_LIMIT = 24; @@ -18,6 +19,10 @@ interface ModelWaiter { ticketId: string; ownerKey: string; resolve: () => void; + onQueued?: (status: CanvasImageModelQueueStatus) => void | Promise; + lastStatusSignature?: string; + debugLabel?: string; + debugMeta?: Record; } interface ModelConcurrencyState { @@ -115,10 +120,6 @@ function normalizeCanvasImageOwnerKey(value: unknown) { return normalized || "__anonymous__"; } -function getMaxConcurrentPerOwner(maxConcurrent: number) { - return Math.max(1, Math.ceil(Math.max(1, maxConcurrent) / 2)); -} - function getRunningCountForOwner( modelState: ModelConcurrencyState, ownerKey: string @@ -132,6 +133,13 @@ function getRunningCountForOwner( return count; } +function hasCompetingWaiter( + modelState: ModelConcurrencyState, + ownerKey: string +) { + return modelState.waiters.some((waiter) => waiter.ownerKey !== ownerKey); +} + export async function getCanvasImageTaskQueueConfig(): Promise { const now = Date.now(); if (cachedConfig && cachedConfig.expiresAt > now) { @@ -236,6 +244,47 @@ function buildModelQueueSnapshot(params: { }; } +function notifyModelWaiters( + modelKey: string, + modelState: ModelConcurrencyState, + concurrencyLimit: number +) { + const sameModelPendingCount = modelState.waiters.length; + for (const [queuePosition, waiter] of modelState.waiters.entries()) { + if (typeof waiter.onQueued !== "function") { + continue; + } + + const queueStatus = buildModelQueueSnapshot({ + modelState, + queuePosition, + sameModelPendingCount, + sameModelRunningCount: modelState.running.size, + concurrencyLimit, + }); + const statusSignature = [ + queueStatus.queuePosition, + queueStatus.sameModelPendingCount, + queueStatus.sameModelRunningCount, + queueStatus.concurrencyLimit, + queueStatus.estimatedWaitMs, + ].join(":"); + if (waiter.lastStatusSignature === statusSignature) { + continue; + } + + waiter.lastStatusSignature = statusSignature; + Promise.resolve(waiter.onQueued(queueStatus)).catch((error) => { + console.warn("[Canvas Image Concurrency] onQueued callback failed:", { + debugLabel: waiter.debugLabel || "canvas-image", + modelKey, + error: error instanceof Error ? error.message : String(error), + ...waiter.debugMeta, + }); + }); + } +} + export async function getCanvasImageModelQueueStatus(params: { modelKey: string; }): Promise { @@ -290,6 +339,7 @@ async function drainModelWaiters( const { concurrency: maxConcurrent, batchDispatchIntervalMs } = await getCanvasImageTaskQueueConfig(); + notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent); if (modelState.running.size >= maxConcurrent) { return; } @@ -314,10 +364,14 @@ async function drainModelWaiters( clearModelDispatchTimer(modelState); while (modelState.running.size < maxConcurrent && modelState.waiters.length > 0) { - const maxConcurrentPerOwner = getMaxConcurrentPerOwner(maxConcurrent); const waiterIndex = modelState.waiters.findIndex( - (item) => - getRunningCountForOwner(modelState, item.ownerKey) < maxConcurrentPerOwner + (item) => { + const ownerLimit = getAdaptiveImageQueueOwnerLimit({ + maxConcurrent, + hasCompetingOwner: hasCompetingWaiter(modelState, item.ownerKey), + }); + return getRunningCountForOwner(modelState, item.ownerKey) < ownerLimit; + } ); if (waiterIndex < 0) { break; @@ -331,6 +385,8 @@ async function drainModelWaiters( waiter.resolve(); } + notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent); + cleanupModelState(normalizedModelKey); } @@ -352,13 +408,16 @@ export async function acquireCanvasImageModelSlot(params: { const waitingSince = Date.now(); const hasActiveDelay = modelState.delayedUntil !== null && modelState.delayedUntil > waitingSince; - const maxConcurrentPerOwner = getMaxConcurrentPerOwner(maxConcurrent); + const ownerLimit = getAdaptiveImageQueueOwnerLimit({ + maxConcurrent, + hasCompetingOwner: hasCompetingWaiter(modelState, normalizedOwnerKey), + }); if ( modelState.waiters.length === 0 && !hasActiveDelay && modelState.running.size < maxConcurrent && - getRunningCountForOwner(modelState, normalizedOwnerKey) < maxConcurrentPerOwner + getRunningCountForOwner(modelState, normalizedOwnerKey) < ownerLimit ) { modelState.running.set(ticketId, normalizedOwnerKey); return { @@ -375,29 +434,15 @@ export async function acquireCanvasImageModelSlot(params: { } await new Promise((resolve) => { - const queuePosition = modelState.waiters.length; modelState.waiters.push({ ticketId, ownerKey: normalizedOwnerKey, resolve, + onQueued: params.onQueued, + debugLabel: params.debugLabel, + debugMeta: params.debugMeta, }); - if (typeof params.onQueued === "function") { - const queueStatus = buildModelQueueSnapshot({ - modelState, - queuePosition, - sameModelPendingCount: modelState.waiters.length, - sameModelRunningCount: modelState.running.size, - concurrencyLimit: maxConcurrent, - }); - Promise.resolve(params.onQueued(queueStatus)).catch((error) => { - console.warn("[Canvas Image Concurrency] onQueued callback failed:", { - debugLabel: params.debugLabel || "canvas-image", - modelKey: normalizedModelKey, - error: error instanceof Error ? error.message : String(error), - ...params.debugMeta, - }); - }); - } + notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent); void drainModelWaiters(normalizedModelKey, "arrival"); }); diff --git a/lib/canvas/image-queue-fairness.ts b/lib/canvas/image-queue-fairness.ts new file mode 100644 index 000000000..41e5b3930 --- /dev/null +++ b/lib/canvas/image-queue-fairness.ts @@ -0,0 +1,11 @@ +export function getAdaptiveImageQueueOwnerLimit(params: { + maxConcurrent: number; + hasCompetingOwner: boolean; +}) { + const maxConcurrent = Math.max(1, Math.floor(params.maxConcurrent || 1)); + if (!params.hasCompetingOwner) { + return maxConcurrent; + } + + return Math.max(1, Math.ceil(maxConcurrent / 2)); +} diff --git a/lib/canvas/image-task-queue.ts b/lib/canvas/image-task-queue.ts index 2edec4d03..a0966ddb4 100644 --- a/lib/canvas/image-task-queue.ts +++ b/lib/canvas/image-task-queue.ts @@ -25,6 +25,7 @@ import { isCanvasLayoutReplaceTaskRequest, type CanvasLayoutReplaceTaskRequest, } from "@/lib/canvas/layout-replace-task"; +import { getAdaptiveImageQueueOwnerLimit } from "./image-queue-fairness"; type CanvasImageTaskStatus = "pending" | "running" | "succeeded" | "failed"; type CanvasImageTaskRequest = @@ -211,10 +212,6 @@ function getModelUserKey(modelKey: string, userId: string) { return `${modelKey}::${userId || "__anonymous__"}`; } -function getMaxConcurrentPerUser(maxConcurrentPerModel: number) { - return Math.max(1, Math.ceil(Math.max(1, maxConcurrentPerModel) / 2)); -} - async function getTaskQueueMeta(taskId: string) { const rows = await db .select({ @@ -358,10 +355,17 @@ async function enqueueLocalTask( state.pendingSet.add(taskId); const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency(); + const hasCompetingUser = state.pending.some( + (entry) => + entry.modelKey === resolvedModelKey && entry.userId !== resolvedUserId + ); + const ownerLimit = getAdaptiveImageQueueOwnerLimit({ + maxConcurrent: maxConcurrentPerModel, + hasCompetingOwner: hasCompetingUser, + }); if ( getRunningCountForModel(resolvedModelKey) >= maxConcurrentPerModel || - getRunningCountForModelUser(resolvedModelKey, resolvedUserId) >= - getMaxConcurrentPerUser(maxConcurrentPerModel) + getRunningCountForModelUser(resolvedModelKey, resolvedUserId) >= ownerLimit ) { markModelAwaitingNextBatch(resolvedModelKey); } @@ -370,12 +374,19 @@ async function enqueueLocalTask( function hasModelCapacity( modelKey: string, userId: string, - maxConcurrentPerModel: number + maxConcurrentPerModel: number, + pendingEntries: PendingTaskEntry[] ) { + const hasCompetingUser = pendingEntries.some( + (entry) => entry.modelKey === modelKey && entry.userId !== userId + ); + const ownerLimit = getAdaptiveImageQueueOwnerLimit({ + maxConcurrent: maxConcurrentPerModel, + hasCompetingOwner: hasCompetingUser, + }); return ( getRunningCountForModel(modelKey) < maxConcurrentPerModel && - getRunningCountForModelUser(modelKey, userId) < - getMaxConcurrentPerUser(maxConcurrentPerModel) + getRunningCountForModelUser(modelKey, userId) < ownerLimit ); } @@ -388,7 +399,14 @@ async function claimNextPendingTask( if (!entry) continue; if (state.running.has(entry.taskId)) continue; if (isModelBatchDispatchDelayed(entry.modelKey)) continue; - if (!hasModelCapacity(entry.modelKey, entry.userId, maxConcurrentPerModel)) { + if ( + !hasModelCapacity( + entry.modelKey, + entry.userId, + maxConcurrentPerModel, + state.pending + ) + ) { markModelAwaitingNextBatch(entry.modelKey); continue; } @@ -414,6 +432,13 @@ async function claimNextPendingTask( .orderBy(asc(canvasImageTask.createdAt)) .limit(PENDING_LOOKAHEAD_LIMIT); + const databasePendingEntries = rows.map((row) => ({ + taskId: row.id, + modelKey: normalizeTaskModelKey(row.request), + userId: String(row.userId || "__anonymous__"), + })); + const capacityCandidates = [...state.pending, ...databasePendingEntries]; + for (const row of rows) { if (!row?.id) continue; if (state.pendingSet.has(row.id) || state.running.has(row.id)) continue; @@ -423,7 +448,14 @@ async function claimNextPendingTask( continue; } const userId = String(row.userId || "__anonymous__"); - if (!hasModelCapacity(modelKey, userId, maxConcurrentPerModel)) { + if ( + !hasModelCapacity( + modelKey, + userId, + maxConcurrentPerModel, + capacityCandidates + ) + ) { markModelAwaitingNextBatch(modelKey); continue; } diff --git a/playwright.helpers.config.ts b/playwright.helpers.config.ts index d3203dfac..9c13d129d 100644 --- a/playwright.helpers.config.ts +++ b/playwright.helpers.config.ts @@ -11,6 +11,7 @@ export default defineConfig({ /routes\/canvas-grouping\.test\.ts/, /routes\/canvas-arrange\.test\.ts/, /routes\/canvas-asset-manager\.test\.ts/, + /routes\/canvas-image-queue-fairness\.test\.ts/, /routes\/canvas-pan\.test\.ts/, /routes\/canvas-zoom\.test\.ts/, /routes\/video-director-project\.test\.ts/, diff --git a/tests/routes/canvas-image-queue-fairness.test.ts b/tests/routes/canvas-image-queue-fairness.test.ts new file mode 100644 index 000000000..c13bac471 --- /dev/null +++ b/tests/routes/canvas-image-queue-fairness.test.ts @@ -0,0 +1,37 @@ +import { expect, test } from "@playwright/test"; +import { getAdaptiveImageQueueOwnerLimit } from "@/lib/canvas/image-queue-fairness"; + +test.describe("canvas image queue fairness", () => { + test("lets one owner use the full model capacity without competition", () => { + expect( + getAdaptiveImageQueueOwnerLimit({ + maxConcurrent: 10, + hasCompetingOwner: false, + }) + ).toBe(10); + }); + + test("reserves half the slots when another owner is waiting", () => { + expect( + getAdaptiveImageQueueOwnerLimit({ + maxConcurrent: 10, + hasCompetingOwner: true, + }) + ).toBe(5); + expect( + getAdaptiveImageQueueOwnerLimit({ + maxConcurrent: 5, + hasCompetingOwner: true, + }) + ).toBe(3); + }); + + test("always keeps at least one executable slot", () => { + expect( + getAdaptiveImageQueueOwnerLimit({ + maxConcurrent: 0, + hasCompetingOwner: true, + }) + ).toBe(1); + }); +});