feat: raise per-model image queue concurrency

app/(admin)/settings/admin-settings-client.tsx

@@ -79,6 +79,7 @@ import {
 import {
   CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY,
   DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
+  MIN_CANVAS_IMAGE_TASK_CONCURRENCY,
   MAX_CANVAS_IMAGE_TASK_CONCURRENCY,
   normalizeCanvasImageTaskQueueConfig,
   type CanvasImageTaskQueueConfig,
@@ -5109,10 +5110,10 @@ export function AdminSettingsClient() {
         normalizeCanvasImageTaskQueueConfig(payload?.setting?.value || normalized)
       );
       await mutateCanvasImageTaskQueueConfig();
-      toast.success("生图任务队列并发配置保存成功");
+      toast.success("生图任务单模型并发配置保存成功");
     } catch (error) {
       toast.error(
-        error instanceof Error ? error.message : "保存生图任务队列并发配置失败"
+        error instanceof Error ? error.message : "保存生图任务单模型并发配置失败"
       );
     }
   };
@@ -20148,16 +20149,16 @@ export function AdminSettingsClient() {
             </CardContent>
           </Card>
 
-          {/* 生图任务队列并发 */}
+          {/* 生图任务单模型并发 */}
           <Card>
             <CardHeader className="flex flex-col gap-4 space-y-0 md:flex-row md:items-start md:justify-between">
               <div className="space-y-1.5">
                 <CardTitle className="flex items-center gap-2 text-xl">
                   <ImageIcon className="h-5 w-5" />
-                  生图任务队列并发
+                  生图任务单模型并发
                 </CardTitle>
                 <CardDescription>
-                  控制风格化/画布异步生图队列允许同时运行的任务数。配置保存到数据库，运行队列会自动读取，无需重启服务。
+                  控制风格化/画布异步生图队列中，同一个模型允许同时运行的任务数。配置保存到数据库，运行队列会自动读取，无需重启服务。
                 </CardDescription>
               </div>
               <LocalAiSectionToggleButton
@@ -20176,19 +20177,20 @@ export function AdminSettingsClient() {
                   <div className="font-medium">生效范围</div>
                   <div className="mt-2 space-y-1 text-xs leading-5 text-amber-800">
                     <p>• 影响通过异步队列提交的风格化/画布生图任务。</p>
-                    <p>• 多用户共享这个全局并发池，不按用户单独拆分。</p>
+                    <p>• 多用户共享队列，但会按模型分别计数，不同模型之间不再互相阻塞。</p>
+                    <p>• 当前配置的是单模型并发，不是全站总并发。</p>
                     <p>• 数值过高可能触发上游模型 429、超时或 504，建议配合模型 Key 池继续细分。</p>
                   </div>
                 </div>
 
                 <div className="space-y-2">
                   <Label htmlFor="canvasImageTaskQueueConcurrency">
-                    最大同时运行任务数
+                    单模型最大同时运行任务数
                   </Label>
                   <Input
                     id="canvasImageTaskQueueConcurrency"
                     type="number"
-                    min={DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}
+                    min={MIN_CANVAS_IMAGE_TASK_CONCURRENCY}
                     max={MAX_CANVAS_IMAGE_TASK_CONCURRENCY}
                     value={canvasImageTaskQueueConfig.concurrency}
                     onChange={(e) =>
@@ -20200,16 +20202,17 @@ export function AdminSettingsClient() {
                     }
                   />
                   <div className="text-xs text-muted-foreground">
-                    可配置范围 {DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}-
+                    可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}-
                     {MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "}
-                    {canvasImageTaskQueueConfig.concurrency} 并发。
+                    {canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "}
+                    {DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。
                   </div>
                 </div>
               </div>
 
               <div className="flex justify-end">
                 <Button onClick={handleSaveCanvasImageTaskQueueConfig}>
-                  <Save className="mr-2 h-4 w-4" /> 保存生图队列并发
+                  <Save className="mr-2 h-4 w-4" /> 保存单模型并发
                 </Button>
               </div>
             </CardContent>

lib/canvas/image-task-config.ts

@@ -1,6 +1,7 @@
 export const CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY = "canvas_image_task_queue_config";
 
-export const DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY = 1;
+export const MIN_CANVAS_IMAGE_TASK_CONCURRENCY = 1;
+export const DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY = 8;
 export const MAX_CANVAS_IMAGE_TASK_CONCURRENCY = 100;
 
 export interface CanvasImageTaskQueueConfig {
@@ -10,7 +11,7 @@ export interface CanvasImageTaskQueueConfig {
 function clampConcurrency(value: number) {
   return Math.min(
     MAX_CANVAS_IMAGE_TASK_CONCURRENCY,
-    Math.max(DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY, Math.floor(value))
+    Math.max(MIN_CANVAS_IMAGE_TASK_CONCURRENCY, Math.floor(value))
   );
 }
 

lib/canvas/image-task-queue.ts

@@ -17,15 +17,22 @@ import {
 
 type CanvasImageTaskStatus = "pending" | "running" | "succeeded" | "failed";
 
+interface PendingTaskEntry {
+  taskId: string;
+  modelKey: string;
+}
+
 interface QueueState {
-  pending: string[];
+  pending: PendingTaskEntry[];
   pendingSet: Set<string>;
-  running: Set<string>;
+  running: Map<string, string>;
+  runningByModel: Map<string, Set<string>>;
   draining: boolean;
 }
 
 const CONCURRENCY_CONFIG_CACHE_MS = 5 * 1000;
 const STALE_RUNNING_MS = 5 * 60 * 1000;
+const PENDING_LOOKAHEAD_LIMIT = 100;
 const queueStateKey = "__canvas_image_task_queue_state__";
 let cachedConcurrency:
   | {
@@ -75,53 +82,138 @@ function getQueueState(): QueueState {
     globalObj[queueStateKey] = {
       pending: [],
       pendingSet: new Set<string>(),
-      running: new Set<string>(),
+      running: new Map<string, string>(),
+      runningByModel: new Map<string, Set<string>>(),
       draining: false,
     };
   }
   return globalObj[queueStateKey] as QueueState;
 }
 
-function enqueueLocalTask(taskId: string) {
+function normalizeTaskModelKey(request: unknown) {
+  if (!request || typeof request !== "object" || Array.isArray(request)) {
+    return "__default__";
+  }
+
+  const rawModel = (request as Record<string, unknown>).model;
+  if (typeof rawModel !== "string") {
+    return "__default__";
+  }
+
+  const normalizedModel = rawModel.trim();
+  return normalizedModel || "__default__";
+}
+
+async function getTaskModelKey(taskId: string) {
+  const rows = await db
+    .select({ request: canvasImageTask.request })
+    .from(canvasImageTask)
+    .where(eq(canvasImageTask.id, taskId))
+    .limit(1);
+
+  return normalizeTaskModelKey(rows[0]?.request);
+}
+
+function getRunningCountForModel(modelKey: string) {
+  const state = getQueueState();
+  return state.runningByModel.get(modelKey)?.size || 0;
+}
+
+function markRunningTask(taskId: string, modelKey: string) {
+  const state = getQueueState();
+  state.running.set(taskId, modelKey);
+  const tasks = state.runningByModel.get(modelKey) || new Set<string>();
+  tasks.add(taskId);
+  state.runningByModel.set(modelKey, tasks);
+}
+
+function releaseRunningTask(taskId: string) {
+  const state = getQueueState();
+  const modelKey = state.running.get(taskId);
+  if (!modelKey) {
+    return;
+  }
+
+  state.running.delete(taskId);
+  const tasks = state.runningByModel.get(modelKey);
+  if (!tasks) {
+    return;
+  }
+
+  tasks.delete(taskId);
+  if (tasks.size === 0) {
+    state.runningByModel.delete(modelKey);
+  }
+}
+
+async function enqueueLocalTask(taskId: string, modelKey?: string) {
   const state = getQueueState();
   if (!taskId || state.running.has(taskId) || state.pendingSet.has(taskId)) {
     return;
   }
-  state.pending.push(taskId);
+
+  const resolvedModelKey = modelKey || (await getTaskModelKey(taskId));
+  state.pending.push({
+    taskId,
+    modelKey: resolvedModelKey,
+  });
   state.pendingSet.add(taskId);
 }
 
-async function claimNextPendingTask(): Promise<string | null> {
+function hasModelCapacity(modelKey: string, maxConcurrentPerModel: number) {
   const state = getQueueState();
-  while (state.pending.length > 0) {
-    const taskId = state.pending.shift();
-    if (!taskId) continue;
-    state.pendingSet.delete(taskId);
-    if (!state.running.has(taskId)) {
-      return taskId;
-    }
-  }
-
-  const runningIds = Array.from(state.running);
-  const whereCondition =
-    runningIds.length > 0
-      ? and(eq(canvasImageTask.status, "pending"), notInArray(canvasImageTask.id, runningIds))
-      : eq(canvasImageTask.status, "pending");
-
-  const rows = await db
-    .select({ id: canvasImageTask.id })
-    .from(canvasImageTask)
-    .where(whereCondition)
-    .orderBy(asc(canvasImageTask.createdAt))
-    .limit(1);
-
-  return rows[0]?.id ?? null;
+  return getRunningCountForModel(modelKey) < maxConcurrentPerModel;
 }
 
-async function runSingleTask(taskId: string) {
+async function claimNextPendingTask(
+  maxConcurrentPerModel: number
+): Promise<PendingTaskEntry | null> {
   const state = getQueueState();
-  state.running.add(taskId);
+  for (let index = 0; index < state.pending.length; index += 1) {
+    const entry = state.pending[index];
+    if (!entry) continue;
+    if (state.running.has(entry.taskId)) continue;
+    if (!hasModelCapacity(entry.modelKey, maxConcurrentPerModel)) continue;
 
+    state.pending.splice(index, 1);
+    state.pendingSet.delete(entry.taskId);
+    return entry;
+  }
+
+  const runningIds = Array.from(state.running.keys());
+  const rows = await db
+    .select({
+      id: canvasImageTask.id,
+      request: canvasImageTask.request,
+    })
+    .from(canvasImageTask)
+    .where(
+      runningIds.length > 0
+        ? and(eq(canvasImageTask.status, "pending"), notInArray(canvasImageTask.id, runningIds))
+        : eq(canvasImageTask.status, "pending")
+    )
+    .orderBy(asc(canvasImageTask.createdAt))
+    .limit(PENDING_LOOKAHEAD_LIMIT);
+
+  for (const row of rows) {
+    if (!row?.id) continue;
+    if (state.pendingSet.has(row.id) || state.running.has(row.id)) continue;
+
+    const modelKey = normalizeTaskModelKey(row.request);
+    if (!hasModelCapacity(modelKey, maxConcurrentPerModel)) {
+      continue;
+    }
+
+    return {
+      taskId: row.id,
+      modelKey,
+    };
+  }
+
+  return null;
+}
+
+async function runSingleTask(entry: PendingTaskEntry) {
   try {
     const [task] = await db
       .update(canvasImageTask)
@@ -133,7 +225,7 @@ async function runSingleTask(taskId: string) {
         error: null,
         attempts: sql`${canvasImageTask.attempts} + 1`,
       })
-      .where(and(eq(canvasImageTask.id, taskId), eq(canvasImageTask.status, "pending")))
+      .where(and(eq(canvasImageTask.id, entry.taskId), eq(canvasImageTask.status, "pending")))
       .returning({
         id: canvasImageTask.id,
         request: canvasImageTask.request,
@@ -153,7 +245,7 @@ async function runSingleTask(taskId: string) {
         completedAt: new Date(),
         updatedAt: new Date(),
       })
-      .where(eq(canvasImageTask.id, taskId));
+      .where(eq(canvasImageTask.id, entry.taskId));
   } catch (error) {
     await db
       .update(canvasImageTask)
@@ -163,9 +255,9 @@ async function runSingleTask(taskId: string) {
         completedAt: new Date(),
         updatedAt: new Date(),
       })
-      .where(eq(canvasImageTask.id, taskId));
+      .where(eq(canvasImageTask.id, entry.taskId));
   } finally {
-    state.running.delete(taskId);
+    releaseRunningTask(entry.taskId);
     void drainCanvasImageTaskQueue();
   }
 }
@@ -176,12 +268,13 @@ export async function drainCanvasImageTaskQueue() {
 
   state.draining = true;
   try {
-    const maxConcurrent = await getQueueConcurrency();
-    while (state.running.size < maxConcurrent) {
-      const taskId = await claimNextPendingTask();
-      if (!taskId) break;
-      if (state.running.has(taskId)) continue;
-      void runSingleTask(taskId);
+    const maxConcurrentPerModel = await getQueueConcurrency();
+    while (true) {
+      const nextTask = await claimNextPendingTask(maxConcurrentPerModel);
+      if (!nextTask) break;
+      if (state.running.has(nextTask.taskId)) continue;
+      markRunningTask(nextTask.taskId, nextTask.modelKey);
+      void runSingleTask(nextTask);
     }
   } finally {
     state.draining = false;
@@ -189,7 +282,7 @@ export async function drainCanvasImageTaskQueue() {
 }
 
 export async function enqueueCanvasImageTask(taskId: string) {
-  enqueueLocalTask(taskId);
+  await enqueueLocalTask(taskId);
   await drainCanvasImageTaskQueue();
 }
 
@@ -256,9 +349,9 @@ export async function getCanvasImageTaskForUser(params: {
             eq(canvasImageTask.id, task.id),
             eq(canvasImageTask.status, "running"),
             lt(canvasImageTask.startedAt, staleBefore)
-          )
+        )
         );
-      enqueueLocalTask(task.id);
+      await enqueueLocalTask(task.id, normalizeTaskModelKey(task.request));
       await drainCanvasImageTaskQueue();
       task.status = "pending" as CanvasImageTaskStatus;
       task.startedAt = null;
@@ -267,7 +360,7 @@ export async function getCanvasImageTaskForUser(params: {
   }
 
   if (task.status === "pending") {
-    enqueueLocalTask(task.id);
+    await enqueueLocalTask(task.id, normalizeTaskModelKey(task.request));
     await drainCanvasImageTaskQueue();
   }