lunanting/ai-chatbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: release image queue slots after dispatch

Browse Source
This commit is contained in:
134 2026-08-03 14:43:21 +08:00
parent 9e6db98195
commit 231be15459
15 changed files with 281 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
app/(chat)/api/chat/_lib/stream.ts
View File

@ -214,6 +214,7 @@ export function writeImageGenerationStatusChunk(
sameModelPendingCount?: number | null;
sameModelRunningCount?: number | null;
concurrencyLimit?: number | null;
dispatchIntervalMs?: number | null;
estimatedWaitMs?: number | null;
},
label: string
@ -249,6 +250,11 @@ export function writeImageGenerationStatusChunk(
Number(params.concurrencyLimit) > 0
? Number(params.concurrencyLimit)
: undefined,
dispatchIntervalMs:
Number.isFinite(Number(params.dispatchIntervalMs)) &&
Number(params.dispatchIntervalMs) >= 0
? Number(params.dispatchIntervalMs)
: undefined,
estimatedWaitMs:
Number.isFinite(Number(params.estimatedWaitMs)) &&
Number(params.estimatedWaitMs) >= 0

8
app/(chat)/api/chat/route.ts
View File

@ -6040,6 +6040,8 @@ export async function POST(request: Request) {
sameModelRunningCount:
queueStatus.sameModelRunningCount,
concurrencyLimit: queueStatus.concurrencyLimit,
dispatchIntervalMs:
queueStatus.dispatchIntervalMs,
estimatedWaitMs: queueStatus.estimatedWaitMs,
},
`agent-mode:detail-replacement:queue:${reference.id}`,
@ -6940,6 +6942,8 @@ export async function POST(request: Request) {
queueStatus.sameModelPendingCount,
sameModelRunningCount: queueStatus.sameModelRunningCount,
concurrencyLimit: queueStatus.concurrencyLimit,
dispatchIntervalMs:
queueStatus.dispatchIntervalMs,
estimatedWaitMs: queueStatus.estimatedWaitMs,
},
`agent-preset:replacement:queue:${reference.id}`,
@ -7267,6 +7271,8 @@ export async function POST(request: Request) {
queueStatus.sameModelPendingCount,
sameModelRunningCount: queueStatus.sameModelRunningCount,
concurrencyLimit: queueStatus.concurrencyLimit,
dispatchIntervalMs:
queueStatus.dispatchIntervalMs,
estimatedWaitMs: queueStatus.estimatedWaitMs,
},
"agent-preset:image:scene-generation:queue",
@ -7496,6 +7502,8 @@ export async function POST(request: Request) {
queueStatus.sameModelPendingCount,
sameModelRunningCount: queueStatus.sameModelRunningCount,
concurrencyLimit: queueStatus.concurrencyLimit,
dispatchIntervalMs:
queueStatus.dispatchIntervalMs,
estimatedWaitMs: queueStatus.estimatedWaitMs,
},
`agent-preset:image:slot-queue:${slot.slotIndex}`,

16
app/api/scene-fusion/route.ts
View File

@ -2134,6 +2134,7 @@ async function processSceneWithDiversity(
sameModelPendingCount: number;
sameModelRunningCount: number;
concurrencyLimit: number;
dispatchIntervalMs: number;
estimatedWaitMs: number;
source: "model_queue";
}) => void | Promise<void>;
@ -2217,12 +2218,13 @@ async function processSceneWithDiversity(
onQueued: params.onQueueStatus,
});
let result: Awaited<ReturnType<typeof processSceneWithProvider>>;
const runtimeModelId = resolveModelRuntimeId(candidate.model, {
providerType: candidate.provider.type,
});
await params.onProcessingStart?.();
let execution: ReturnType<typeof processSceneWithProvider> | null = null;
try {
const runtimeModelId = resolveModelRuntimeId(candidate.model, {
providerType: candidate.provider.type,
});
await params.onProcessingStart?.();
result = await processSceneWithProvider(
execution = processSceneWithProvider(
params.productImageUrl,
params.sceneImageUrl,
params.productName,
@ -2246,6 +2248,10 @@ async function processSceneWithDiversity(
} finally {
await modelSlot.release();
}
if (!execution) {
throw new Error("场景裂变请求未能提交到模型服务");
}
result = await execution;
const echoCheck = await detectEchoedProductResult({
requestId: params.requestId,

10
components/canvas/CanvasChatPanel.tsx
View File

@ -7190,6 +7190,11 @@ export function CanvasChatPanel({
Number(payload.concurrencyLimit) > 0
? Number(payload.concurrencyLimit)
: undefined,
dispatchIntervalMs:
Number.isFinite(Number(payload.dispatchIntervalMs)) &&
Number(payload.dispatchIntervalMs) >= 0
? Number(payload.dispatchIntervalMs)
: undefined,
estimatedWaitMs:
Number.isFinite(Number(payload.estimatedWaitMs)) &&
Number(payload.estimatedWaitMs) >= 0
@ -13699,6 +13704,8 @@ export function CanvasChatPanel({
const loadingSameModelRunningCount =
streamImageGenerationState?.sameModelRunningCount;
const loadingConcurrencyLimit = streamImageGenerationState?.concurrencyLimit;
const loadingDispatchIntervalMs =
streamImageGenerationState?.dispatchIntervalMs;
const loadingEstimatedWaitMs = streamImageGenerationState?.estimatedWaitMs;
const canvasGenerationPlaceholderState =
useMemo<ImageGenerationStatusStreamPayload | null>(() => {
@ -13714,10 +13721,12 @@ export function CanvasChatPanel({
sameModelPendingCount: loadingSameModelPendingCount,
sameModelRunningCount: loadingSameModelRunningCount,
concurrencyLimit: loadingConcurrencyLimit,
dispatchIntervalMs: loadingDispatchIntervalMs,
estimatedWaitMs: loadingEstimatedWaitMs,
};
}, [
loadingConcurrencyLimit,
loadingDispatchIntervalMs,
loadingEstimatedWaitMs,
loadingImageCount,
loadingModelName,
@ -15451,6 +15460,7 @@ export function CanvasChatPanel({
sameModelPendingCount={loadingSameModelPendingCount}
sameModelRunningCount={loadingSameModelRunningCount}
concurrencyLimit={loadingConcurrencyLimit}
dispatchIntervalMs={loadingDispatchIntervalMs}
estimatedWaitMs={loadingEstimatedWaitMs}
/>
)}

1
components/canvas/chat-panel/canvas-agent-stream.ts
View File

@ -158,6 +158,7 @@ export type ImageGenerationStatusStreamPayload = {
sameModelPendingCount?: number;
sameModelRunningCount?: number;
concurrencyLimit?: number;
dispatchIntervalMs?: number;
estimatedWaitMs?: number;
};

13
components/image-generation-loading.tsx
View File

@ -18,6 +18,7 @@ interface ImageGenerationLoadingProps {
sameModelPendingCount?: number;
sameModelRunningCount?: number;
concurrencyLimit?: number;
dispatchIntervalMs?: number;
estimatedWaitMs?: number;
}
@ -110,7 +111,7 @@ export const ImageGenerationLoading = memo(
queuePosition,
sameModelPendingCount,
sameModelRunningCount,
concurrencyLimit,
dispatchIntervalMs,
estimatedWaitMs,
}: ImageGenerationLoadingProps) => {
const shouldTrackElapsed =
@ -164,10 +165,16 @@ export const ImageGenerationLoading = memo(
? `队列共 ${Math.max(
0,
Number(sameModelPendingCount || 0)
)} 个待处理，当前模型运行 ${Math.max(
)} 个待提交，已提交 ${Math.max(
0,
Number(sameModelRunningCount || 0)
)}/${Math.max(1, Number(concurrencyLimit || 0) || 1)}，${formatImageQueueWaitLabel(
)} 个生成任务，${
Number(dispatchIntervalMs || 0) > 0
? `每 ${Math.max(0.1, Number(dispatchIntervalMs) / 1000).toFixed(
Number(dispatchIntervalMs) % 1000 === 0 ? 0 : 1
)} 秒提交 1 个，`
: ""
}${formatImageQueueWaitLabel(
Math.max(
0,
Number(estimatedWaitMs || 0) - resolvedElapsedSeconds * 1000

21
components/scene-fusion-view.tsx
View File

@ -114,6 +114,7 @@ interface ProcessItem {
sameModelPendingCount?: number;
sameModelRunningCount?: number;
concurrencyLimit?: number;
dispatchIntervalMs?: number;
estimatedWaitMs?: number;
source?: "model_queue" | "batch_queue";
};
@ -3067,10 +3068,6 @@ export function SceneFusionView({
0,
Number(item.queueStatus?.sameModelRunningCount || 0)
);
const concurrencyLimit = Math.max(
0,
Number(item.queueStatus?.concurrencyLimit || 0)
);
const serverQueuePosition = Math.max(
0,
Number(item.queueStatus?.queuePosition || 0)
@ -3094,11 +3091,11 @@ export function SceneFusionView({
const queueDetail =
item.status === "pending"
? serverQueuePosition > 0
? `当前模型运行 ${serverRunningCount}/${Math.max(
concurrencyLimit,
serverRunningCount,
1
)}，${formatQueueWaitLabel(
? `前方任务将按 ${Math.max(
0.1,
Number(item.queueStatus?.dispatchIntervalMs || 0) /
1000
)} 秒间隔提交，${formatQueueWaitLabel(
item.queueStatus?.estimatedWaitMs
)}`
: localBatchQueueMeta
@ -3106,11 +3103,7 @@ export function SceneFusionView({
: "点击开始批量处理后进入队列"
: item.status === "processing" &&
serverQueuePosition > 0
? `当前模型运行 ${serverRunningCount}/${Math.max(
concurrencyLimit,
serverRunningCount,
1
)}，${formatQueueWaitLabel(
? `已提交 ${serverRunningCount} 个生成任务，${formatQueueWaitLabel(
item.queueStatus?.estimatedWaitMs
)}`
: null;

16
components/style-pro/StyleProWorkspace.tsx
View File

@ -112,6 +112,7 @@ type StyleProGenerateResponse = {
sameModelPendingCount?: number;
sameModelRunningCount?: number;
concurrencyLimit?: number;
dispatchIntervalMs?: number;
estimatedWaitMs?: number;
source?: "model_queue" | "batch_queue";
};
@ -709,10 +710,6 @@ function GeneratingImagePlaceholder({
0,
Number(queueStatus?.sameModelRunningCount || 0)
);
const concurrencyLimit = Math.max(
0,
Number(queueStatus?.concurrencyLimit || 0)
);
const isQueued = taskState?.status === "pending";
const isFailed = taskState?.status === "failed";
const title = isFailed
@ -725,9 +722,14 @@ function GeneratingImagePlaceholder({
const description = isFailed
? taskState?.error || "请调整参数后重试。"
: isQueued
? `当前模型运行 ${sameModelRunningCount}/${concurrencyLimit || "?"}，${formatStyleProEstimatedWait(
queueStatus?.estimatedWaitMs
)}`
? `已提交 ${sameModelRunningCount} 个生成任务，${
Number(queueStatus?.dispatchIntervalMs || 0) > 0
? `每 ${Math.max(
0.1,
Number(queueStatus?.dispatchIntervalMs || 0) / 1000
)} 秒提交 1 个，`
: ""
}${formatStyleProEstimatedWait(queueStatus?.estimatedWaitMs)}`
: taskState?.generationMode === "realistic"
? "系统会先完成基础实景化，再合成最终风格结果。"
: "模型已开始执行，结果完成后会自动回填。";

110
lib/canvas/generate-image.ts
View File

@ -139,6 +139,7 @@ export interface CanvasGenerateImageRequest {
modelRef?: string | null;
}
) => void | Promise<void>;
onRequestDispatched?: () => void | Promise<void>;
}
export interface ParsedImage {
@ -4256,41 +4257,9 @@ export async function generateCanvasImage(
const resolvedModelKey = normalizeCanvasImageModelKey(
selectedModel.model.modelId || selectedModel.model.id || model
);
const slot = await acquireCanvasImageModelSlot({
modelKey: resolvedModelKey,
ownerKey,
debugLabel,
debugMeta: {
selectedModelId: selectedModel.model.modelId,
selectedModelRef: selectedModel.model.id,
providerType: provider.type,
...debugMeta,
},
onQueued: request.onQueueStatusChange
? async (queueStatus) => {
await request.onQueueStatusChange?.({
...queueStatus,
modelName: selectedModel.model.displayName || null,
modelRef: selectedModel.model.id || null,
});
}
: undefined,
});
if (slot.waitMs > 0 && request.onQueueStatusChange) {
const runningStatus = await getCanvasImageModelQueueStatus({
modelKey: resolvedModelKey,
});
await request.onQueueStatusChange({
...runningStatus,
status: "running",
queuePosition: 0,
estimatedWaitMs: 0,
modelName: selectedModel.model.displayName || null,
modelRef: selectedModel.model.id || null,
});
}
let totalQueueWaitMs = 0;
let singleModelConcurrency = 1;
let requestDispatchNotified = false;
let caughtError: unknown = null;
try {
@ -4298,6 +4267,44 @@ export async function generateCanvasImage(
const remaining = targetImageCount - collectedImages.length;
if (remaining <= 0) break;
const batchImageCount = Math.min(providerBatchLimit, remaining);
const slot = await acquireCanvasImageModelSlot({
modelKey: resolvedModelKey,
ownerKey,
debugLabel,
debugMeta: {
selectedModelId: selectedModel.model.modelId,
selectedModelRef: selectedModel.model.id,
providerType: provider.type,
batchIndex: batchIndex + 1,
...debugMeta,
},
onQueued: request.onQueueStatusChange
? async (queueStatus) => {
await request.onQueueStatusChange?.({
...queueStatus,
modelName: selectedModel.model.displayName || null,
modelRef: selectedModel.model.id || null,
});
}
: undefined,
});
totalQueueWaitMs += slot.waitMs;
singleModelConcurrency = slot.maxConcurrent;
if (slot.waitMs > 0 && request.onQueueStatusChange) {
const runningStatus = await getCanvasImageModelQueueStatus({
modelKey: resolvedModelKey,
});
await request.onQueueStatusChange({
...runningStatus,
status: "running",
queuePosition: 0,
estimatedWaitMs: 0,
modelName: selectedModel.model.displayName || null,
modelRef: selectedModel.model.id || null,
});
}
const batchStartedAt = Date.now();
console.info("[Canvas Generate Image] Batch started:", {
debugLabel,
@ -4309,8 +4316,29 @@ export async function generateCanvasImage(
selectedModelId: selectedModel.model.modelId,
...debugMeta,
});
let batchExecution: Promise<string>;
try {
batchExecution = executeGenerateBatch(batchImageCount);
} finally {
await slot.release();
}
if (!requestDispatchNotified) {
requestDispatchNotified = true;
await Promise.resolve(request.onRequestDispatched?.()).catch(
(dispatchCallbackError) => {
console.warn("[Canvas Generate Image] Dispatch callback failed:", {
debugLabel,
error:
dispatchCallbackError instanceof Error
? dispatchCallbackError.message
: String(dispatchCallbackError),
...debugMeta,
});
}
);
}
const imageMarkdown = await withCanvasProviderBatchTimeout(
executeGenerateBatch(batchImageCount),
batchExecution,
{
debugLabel,
modelId: selectedModel.model.modelId,
@ -4362,8 +4390,8 @@ export async function generateCanvasImage(
console.info("[Canvas Generate Image] Completed:", {
debugLabel,
elapsedMs: Date.now() - startedAt,
waitMs: slot.waitMs,
singleModelConcurrency: slot.maxConcurrent,
waitMs: totalQueueWaitMs,
singleModelConcurrency,
finalImageCount: finalImages.length,
providerType: provider.type,
selectedModelRef: selectedModel.model.id,
@ -4396,8 +4424,8 @@ export async function generateCanvasImage(
console.error("[Canvas Generate Image] Failed:", {
debugLabel,
elapsedMs: Date.now() - startedAt,
waitMs: slot.waitMs,
singleModelConcurrency: slot.maxConcurrent,
waitMs: totalQueueWaitMs,
singleModelConcurrency,
providerType: provider.type,
selectedModelRef: selectedModel.model.id,
selectedModelId: selectedModel.model.modelId,
@ -4405,8 +4433,6 @@ export async function generateCanvasImage(
...debugMeta,
});
caughtError = error;
} finally {
await slot.release();
}
const retryVariants = Array.isArray(request.retryVariants)

100
lib/canvas/image-concurrency.ts
View File

@ -8,7 +8,10 @@ import {
type CanvasImageTaskQueueConfig,
normalizeCanvasImageTaskQueueConfig,
} from "./image-task-config";
import { getAdaptiveImageQueueOwnerLimit } from "./image-queue-fairness";
import {
estimateImageDispatchWaitMs,
getAdaptiveImageQueueOwnerLimit,
} from "./image-queue-fairness";
const CONCURRENCY_CONFIG_CACHE_MS = 5 * 1000;
const RECENT_EXECUTION_SAMPLE_LIMIT = 24;
@ -95,7 +98,9 @@ function cleanupModelState(modelKey: string) {
if (
modelState.running.size === 0 &&
modelState.waiters.length === 0 &&
modelState.recentExecutionDurationsMs.length === 0
modelState.recentExecutionDurationsMs.length === 0 &&
!modelState.dispatchTimer &&
(!modelState.delayedUntil || modelState.delayedUntil <= Date.now())
) {
clearModelDispatchTimer(modelState);
state.models.delete(modelKey);
@ -201,6 +206,7 @@ export type CanvasImageModelQueueStatus = {
sameModelPendingCount: number;
sameModelRunningCount: number;
concurrencyLimit: number;
dispatchIntervalMs: number;
estimatedWaitMs: number;
source: "model_queue";
};
@ -222,16 +228,28 @@ function buildModelQueueSnapshot(params: {
sameModelPendingCount: number;
sameModelRunningCount: number;
concurrencyLimit: number;
dispatchIntervalMs: number;
}): CanvasImageModelQueueStatus {
const averageExecutionMs = getRecentAverageExecutionMs(params.modelState);
const delayRemainingMs = Math.max(
0,
Number(params.modelState.delayedUntil || 0) - Date.now()
);
const estimatedWaitMs =
params.sameModelPendingCount <= 0
? 0
: Math.max(
averageExecutionMs,
Math.ceil((params.queuePosition + 1) / Math.max(1, params.concurrencyLimit)) *
averageExecutionMs
);
: params.dispatchIntervalMs > 0
? estimateImageDispatchWaitMs({
queuePosition: params.queuePosition,
dispatchIntervalMs: params.dispatchIntervalMs,
delayRemainingMs,
})
: Math.max(
averageExecutionMs,
Math.ceil(
(params.queuePosition + 1) / Math.max(1, params.concurrencyLimit)
) * averageExecutionMs
);
return {
status: "pending",
@ -239,6 +257,7 @@ function buildModelQueueSnapshot(params: {
sameModelPendingCount: Math.max(0, params.sameModelPendingCount),
sameModelRunningCount: Math.max(0, params.sameModelRunningCount),
concurrencyLimit: Math.max(1, params.concurrencyLimit),
dispatchIntervalMs: Math.max(0, params.dispatchIntervalMs),
estimatedWaitMs,
source: "model_queue",
};
@ -247,7 +266,8 @@ function buildModelQueueSnapshot(params: {
function notifyModelWaiters(
modelKey: string,
modelState: ModelConcurrencyState,
concurrencyLimit: number
concurrencyLimit: number,
dispatchIntervalMs: number
) {
const sameModelPendingCount = modelState.waiters.length;
for (const [queuePosition, waiter] of modelState.waiters.entries()) {
@ -261,12 +281,14 @@ function notifyModelWaiters(
sameModelPendingCount,
sameModelRunningCount: modelState.running.size,
concurrencyLimit,
dispatchIntervalMs,
});
const statusSignature = [
queueStatus.queuePosition,
queueStatus.sameModelPendingCount,
queueStatus.sameModelRunningCount,
queueStatus.concurrencyLimit,
queueStatus.dispatchIntervalMs,
queueStatus.estimatedWaitMs,
].join(":");
if (waiter.lastStatusSignature === statusSignature) {
@ -289,7 +311,8 @@ export async function getCanvasImageModelQueueStatus(params: {
modelKey: string;
}): Promise<CanvasImageModelQueueStatus> {
const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey);
const { concurrency: maxConcurrent } = await getCanvasImageTaskQueueConfig();
const { concurrency: maxConcurrent, batchDispatchIntervalMs } =
await getCanvasImageTaskQueueConfig();
const modelState =
getGlobalConcurrencyState().models.get(normalizedModelKey) || {
running: new Map<string, string>(),
@ -304,6 +327,7 @@ export async function getCanvasImageModelQueueStatus(params: {
sameModelPendingCount: modelState.waiters.length,
sameModelRunningCount: modelState.running.size,
concurrencyLimit: maxConcurrent,
dispatchIntervalMs: batchDispatchIntervalMs,
});
}
@ -328,7 +352,7 @@ export function recordCanvasImageModelExecutionDuration(
async function drainModelWaiters(
modelKey: string,
trigger: "arrival" | "slot-release" | "timer"
_trigger: "arrival" | "slot-release" | "timer"
) {
const normalizedModelKey = normalizeCanvasImageModelKey(modelKey);
const modelState = getGlobalConcurrencyState().models.get(normalizedModelKey);
@ -339,7 +363,12 @@ async function drainModelWaiters(
const { concurrency: maxConcurrent, batchDispatchIntervalMs } =
await getCanvasImageTaskQueueConfig();
notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent);
notifyModelWaiters(
normalizedModelKey,
modelState,
maxConcurrent,
batchDispatchIntervalMs
);
if (modelState.running.size >= maxConcurrent) {
return;
}
@ -349,19 +378,6 @@ async function drainModelWaiters(
return;
}
if (trigger === "slot-release" && batchDispatchIntervalMs > 0) {
if (!modelState.dispatchTimer) {
modelState.delayedUntil = now + batchDispatchIntervalMs;
modelState.dispatchTimer = setTimeout(() => {
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
if (!state) return;
clearModelDispatchTimer(state);
void drainModelWaiters(normalizedModelKey, "timer");
}, batchDispatchIntervalMs);
}
return;
}
clearModelDispatchTimer(modelState);
while (modelState.running.size < maxConcurrent && modelState.waiters.length > 0) {
const waiterIndex = modelState.waiters.findIndex(
@ -383,9 +399,24 @@ async function drainModelWaiters(
}
modelState.running.set(waiter.ticketId, waiter.ownerKey);
waiter.resolve();
if (batchDispatchIntervalMs > 0) {
modelState.delayedUntil = Date.now() + batchDispatchIntervalMs;
modelState.dispatchTimer = setTimeout(() => {
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
if (!state) return;
clearModelDispatchTimer(state);
void drainModelWaiters(normalizedModelKey, "timer");
}, batchDispatchIntervalMs);
break;
}
}
notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent);
notifyModelWaiters(
normalizedModelKey,
modelState,
maxConcurrent,
batchDispatchIntervalMs
);
cleanupModelState(normalizedModelKey);
}
@ -408,6 +439,9 @@ export async function acquireCanvasImageModelSlot(params: {
const waitingSince = Date.now();
const hasActiveDelay =
modelState.delayedUntil !== null && modelState.delayedUntil > waitingSince;
if (!hasActiveDelay && modelState.dispatchTimer) {
clearModelDispatchTimer(modelState);
}
const ownerLimit = getAdaptiveImageQueueOwnerLimit({
maxConcurrent,
hasCompetingOwner: hasCompetingWaiter(modelState, normalizedOwnerKey),
@ -420,6 +454,15 @@ export async function acquireCanvasImageModelSlot(params: {
getRunningCountForOwner(modelState, normalizedOwnerKey) < ownerLimit
) {
modelState.running.set(ticketId, normalizedOwnerKey);
if (batchDispatchIntervalMs > 0) {
modelState.delayedUntil = Date.now() + batchDispatchIntervalMs;
modelState.dispatchTimer = setTimeout(() => {
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
if (!state) return;
clearModelDispatchTimer(state);
void drainModelWaiters(normalizedModelKey, "timer");
}, batchDispatchIntervalMs);
}
return {
release: async () => {
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
@ -442,7 +485,12 @@ export async function acquireCanvasImageModelSlot(params: {
debugLabel: params.debugLabel,
debugMeta: params.debugMeta,
});
notifyModelWaiters(normalizedModelKey, modelState, maxConcurrent);
notifyModelWaiters(
normalizedModelKey,
modelState,
maxConcurrent,
batchDispatchIntervalMs
);
void drainModelWaiters(normalizedModelKey, "arrival");
});

11
lib/canvas/image-queue-fairness.ts
View File

@ -9,3 +9,14 @@ export function getAdaptiveImageQueueOwnerLimit(params: {
return Math.max(1, Math.ceil(maxConcurrent / 2));
}
export function estimateImageDispatchWaitMs(params: {
queuePosition: number;
dispatchIntervalMs: number;
delayRemainingMs?: number;
}) {
const queuePosition = Math.max(0, Math.floor(params.queuePosition || 0));
const dispatchIntervalMs = Math.max(0, params.dispatchIntervalMs || 0);
const delayRemainingMs = Math.max(0, params.delayRemainingMs || 0);
return delayRemainingMs + queuePosition * dispatchIntervalMs;
}

107
lib/canvas/image-task-queue.ts
View File

@ -4,9 +4,9 @@ import { and, asc, desc, eq, lt, notInArray, sql } from "drizzle-orm";
import { db } from "@/lib/db";
import { canvasImageTask } from "@/lib/db/schema";
import {
getCanvasImageBatchDispatchIntervalMs,
getCanvasImageRunningCountForModel,
getCanvasImageSingleModelConcurrency,
getCanvasImageTaskQueueConfig,
normalizeCanvasImageModelKey,
} from "./image-concurrency";
import {
@ -25,7 +25,10 @@ import {
isCanvasLayoutReplaceTaskRequest,
type CanvasLayoutReplaceTaskRequest,
} from "@/lib/canvas/layout-replace-task";
import { getAdaptiveImageQueueOwnerLimit } from "./image-queue-fairness";
import {
estimateImageDispatchWaitMs,
getAdaptiveImageQueueOwnerLimit,
} from "./image-queue-fairness";
type CanvasImageTaskStatus = "pending" | "running" | "succeeded" | "failed";
type CanvasImageTaskRequest =
@ -47,6 +50,7 @@ export type CanvasImageTaskQueueStatus = {
sameModelPendingCount: number;
sameModelRunningCount: number;
concurrencyLimit: number;
dispatchIntervalMs: number;
estimatedWaitMs: number;
source: "model_queue" | "batch_queue";
};
@ -98,8 +102,6 @@ const LOCAL_ACTIVE_RUNNING_HARD_TIMEOUT_MS = 30 * 60 * 1000;
const STALE_RUNNING_CLEANUP_INTERVAL_MS = 60 * 1000;
const PENDING_LOOKAHEAD_LIMIT = 100;
const QUEUE_STATUS_PENDING_LOOKAHEAD_LIMIT = 500;
const QUEUE_STATUS_RECENT_SAMPLE_LIMIT = 120;
const DEFAULT_ESTIMATED_WAIT_MS = 45 * 1000;
const queueStateKey = "__canvas_image_task_queue_state__";
function getQueueState(): QueueState {
@ -513,6 +515,16 @@ async function markStaleDatabaseRunningTasksFailed() {
}
async function runSingleTask(entry: PendingTaskEntry) {
let dispatchSlotReleased = false;
const releaseDispatchSlot = async () => {
if (dispatchSlotReleased) return;
dispatchSlotReleased = true;
releaseRunningTask(entry.taskId);
setTimeout(() => {
void drainCanvasImageTaskQueue();
}, 0);
};
try {
const [task] = await db
.update(canvasImageTask)
@ -541,15 +553,18 @@ async function runSingleTask(entry: PendingTaskEntry) {
? executeStyleProTask({
request,
userId: String(task.userId || entry.userId),
onRequestDispatched: releaseDispatchSlot,
})
: isCanvasLayoutReplaceTaskRequest(request)
? executeCanvasLayoutReplaceTask({
request,
userId: String(task.userId || entry.userId),
onRequestDispatched: releaseDispatchSlot,
})
: generateCanvasImage({
...(request as CanvasGenerateImageRequest),
ownerKey: task.userId,
onRequestDispatched: releaseDispatchSlot,
});
const result = await withCanvasImageTaskExecutionTimeout(
execution,
@ -581,14 +596,7 @@ async function runSingleTask(entry: PendingTaskEntry) {
and(eq(canvasImageTask.id, entry.taskId), eq(canvasImageTask.status, "running"))
);
} finally {
const modelKey = releaseRunningTask(entry.taskId);
if (modelKey) {
const batchDispatchIntervalMs = await getCanvasImageBatchDispatchIntervalMs();
if (getQueueState().awaitingNextBatch.has(modelKey)) {
scheduleModelNextBatch(modelKey, batchDispatchIntervalMs);
}
}
void drainCanvasImageTaskQueue();
await releaseDispatchSlot();
}
}
@ -599,12 +607,14 @@ export async function drainCanvasImageTaskQueue() {
state.draining = true;
try {
await markStaleDatabaseRunningTasksFailed();
const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency();
const { concurrency: maxConcurrentPerModel, batchDispatchIntervalMs } =
await getCanvasImageTaskQueueConfig();
while (true) {
const nextTask = await claimNextPendingTask(maxConcurrentPerModel);
if (!nextTask) break;
if (state.running.has(nextTask.taskId)) continue;
markRunningTask(nextTask.taskId, nextTask.modelKey, nextTask.userId);
scheduleModelNextBatch(nextTask.modelKey, batchDispatchIntervalMs);
void runSingleTask(nextTask);
}
} finally {
@ -670,6 +680,7 @@ export async function generateCanvasImageWithPersistentQueue(params: {
}): Promise<CanvasGenerateImageResult> {
const {
onQueueStatusChange,
onRequestDispatched: _requestDispatchCallback,
ownerKey: _requestOwnerKey,
...serializableRequest
} = params.request;
@ -700,6 +711,7 @@ export async function generateCanvasImageWithPersistentQueue(params: {
queueStatus.sameModelPendingCount,
queueStatus.sameModelRunningCount,
queueStatus.concurrencyLimit,
queueStatus.dispatchIntervalMs,
].join(":");
if (statusSignature !== lastStatusSignature) {
lastStatusSignature = statusSignature;
@ -739,58 +751,6 @@ export async function generateCanvasImageWithPersistentQueue(params: {
throw new Error(`图片生成任务等待超时（${task.id}）`);
}
async function estimateModelQueueWaitMs(params: {
modelKey: string;
sameModelRunningCount: number;
queuePosition: number;
concurrencyLimit: number;
}) {
const rows = await db
.select({
request: canvasImageTask.request,
startedAt: canvasImageTask.startedAt,
completedAt: canvasImageTask.completedAt,
})
.from(canvasImageTask)
.where(eq(canvasImageTask.status, "succeeded"))
.orderBy(desc(canvasImageTask.completedAt))
.limit(QUEUE_STATUS_RECENT_SAMPLE_LIMIT);
const durations = rows
.filter((row) => normalizeTaskModelKey(row.request) === params.modelKey)
.map((row) => {
if (!row.startedAt || !row.completedAt) return 0;
return row.completedAt.getTime() - row.startedAt.getTime();
})
.filter((value) => Number.isFinite(value) && value > 1000)
.slice(0, 12);
const averageDurationMs =
durations.length > 0
? Math.round(durations.reduce((sum, value) => sum + value, 0) / durations.length)
: DEFAULT_ESTIMATED_WAIT_MS;
const immediatelyStartable = Math.max(
params.concurrencyLimit - params.sameModelRunningCount,
0
);
if (params.queuePosition < immediatelyStartable) {
return 0;
}
const tasksAheadAfterImmediateSlots = Math.max(
0,
params.queuePosition - immediatelyStartable
);
const waveCount =
Math.floor(
tasksAheadAfterImmediateSlots / Math.max(1, params.concurrencyLimit)
) +
(params.sameModelRunningCount > 0 || immediatelyStartable === 0 ? 1 : 0);
return Math.max(0, waveCount * averageDurationMs);
}
async function buildCanvasImageTaskQueueStatus(task: {
id: string;
status: CanvasImageTaskStatus;
@ -798,7 +758,10 @@ async function buildCanvasImageTaskQueueStatus(task: {
createdAt: Date;
}) {
const modelKey = normalizeTaskModelKey(task.request);
const concurrencyLimit = await getCanvasImageSingleModelConcurrency();
const {
concurrency: concurrencyLimit,
batchDispatchIntervalMs: dispatchIntervalMs,
} = await getCanvasImageTaskQueueConfig();
const runningRows = await db
.select({
request: canvasImageTask.request,
@ -817,6 +780,7 @@ async function buildCanvasImageTaskQueueStatus(task: {
sameModelPendingCount: 0,
sameModelRunningCount,
concurrencyLimit,
dispatchIntervalMs,
estimatedWaitMs: 0,
source: "model_queue" as const,
};
@ -841,11 +805,13 @@ async function buildCanvasImageTaskQueueStatus(task: {
0
);
const sameModelPendingCount = sameModelPendingRows.length;
const estimatedWaitMs = await estimateModelQueueWaitMs({
modelKey,
sameModelRunningCount,
const estimatedWaitMs = estimateImageDispatchWaitMs({
queuePosition,
concurrencyLimit,
dispatchIntervalMs,
delayRemainingMs: Math.max(
0,
Number(getQueueState().nextBatchEligibleAt.get(modelKey) || 0) - Date.now()
),
});
return {
@ -854,6 +820,7 @@ async function buildCanvasImageTaskQueueStatus(task: {
sameModelPendingCount,
sameModelRunningCount,
concurrencyLimit,
dispatchIntervalMs,
estimatedWaitMs,
source: "model_queue" as const,
};

2
lib/canvas/layout-replace-task.ts
View File

@ -466,6 +466,7 @@ async function buildLayoutLockedCompositeImage(params: {
export async function executeCanvasLayoutReplaceTask(params: {
request: CanvasLayoutReplaceTaskRequest;
userId: string;
onRequestDispatched?: () => void | Promise<void>;
}): Promise<CanvasGenerateImageResult> {
const sourceImageUrl = normalizeHttpUrl(params.request.sourceImageUrl);
const referenceImageUrl = normalizeHttpUrl(params.request.referenceImageUrl);
@ -542,6 +543,7 @@ export async function executeCanvasLayoutReplaceTask(params: {
displayPrompt: "排版替换结果",
model: resolvedModelRef,
ownerKey: params.userId,
onRequestDispatched: params.onRequestDispatched,
aspectRatio: referenceGeometry?.aspectRatio || "auto",
imageSize: params.request.imageSize || "2K",
imageCount: 1,

4
lib/style-pro/task.ts
View File

@ -264,6 +264,7 @@ async function saveStyleProFinalImageToGallery(params: {
export async function executeStyleProTask(params: {
request: StyleProTaskRequest;
userId: string;
onRequestDispatched?: () => void | Promise<void>;
}) {
const { request, userId } = params;
const aspectRatio = request.prompts.aspectRatio || "auto";
@ -289,6 +290,7 @@ export async function executeStyleProTask(params: {
imageCount: 1,
referenceImage: request.sourceImageUrl,
ownerKey: userId,
onRequestDispatched: params.onRequestDispatched,
allowFailover: false,
debugLabel: "style-pro-realistic-stage1",
debugMeta: {
@ -317,6 +319,7 @@ export async function executeStyleProTask(params: {
imageCount: 1,
referenceImage: firstPassImage.imageUrl,
ownerKey: userId,
onRequestDispatched: params.onRequestDispatched,
allowFailover: false,
debugLabel: "style-pro-realistic-stage2",
debugMeta: {
@ -342,6 +345,7 @@ export async function executeStyleProTask(params: {
imageCount: 1,
referenceImage: request.sourceImageUrl,
ownerKey: userId,
onRequestDispatched: params.onRequestDispatched,
allowFailover: false,
debugLabel: "style-pro",
debugMeta: {

24
tests/routes/canvas-image-queue-fairness.test.ts
View File

@ -1,5 +1,8 @@
import { expect, test } from "@playwright/test";
import { getAdaptiveImageQueueOwnerLimit } from "@/lib/canvas/image-queue-fairness";
import {
estimateImageDispatchWaitMs,
getAdaptiveImageQueueOwnerLimit,
} from "@/lib/canvas/image-queue-fairness";
test.describe("canvas image queue fairness", () => {
test("lets one owner use the full model capacity without competition", () => {
@ -34,4 +37,23 @@ test.describe("canvas image queue fairness", () => {
})
).toBe(1);
});
test("estimates queue wait from request dispatch spacing, not generation duration", () => {
expect(
estimateImageDispatchWaitMs({
queuePosition: 3,
dispatchIntervalMs: 1000,
delayRemainingMs: 400,
})
).toBe(3400);
});
test("does not add wait after the request has no tasks ahead", () => {
expect(
estimateImageDispatchWaitMs({
queuePosition: 0,
dispatchIntervalMs: 1000,
})
).toBe(0);
});
});