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

fix(queue): improve image throughput and live status

Browse Source
This commit is contained in:
134 2026-08-03 13:47:56 +08:00
parent 5e74c44430
commit 9e6db98195
10 changed files with 208 additions and 40 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

@ -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

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

@ -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,

10
components/canvas/CanvasChatPanel.tsx
View File

@ -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}

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

@ -155,6 +155,7 @@ export type ImageGenerationStatusStreamPayload = {
modelName?: string;
imageCount?: number;
queuePosition?: number;
sameModelPendingCount?: number;
sameModelRunningCount?: number;
concurrencyLimit?: number;
estimatedWaitMs?: number;

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

@ -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 (
<div
@ -198,6 +212,9 @@ export const ImageGenerationLoading = memo(
</div>
{queueLine ? <div className="text-[13px]">{queueLine}</div> : null}
{queueDetailLine ? <div className="text-[13px]">{queueDetailLine}</div> : null}
{queueTimeoutHint ? (
<div className="text-[13px] text-amber-500">{queueTimeoutHint}</div>
) : null}
{(phase !== "processing" || resolvedElapsedSeconds > 0) && (
<div className="text-[13px]">
{phaseHint} {elapsedLabel}

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

@ -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<void>;
lastStatusSignature?: string;
debugLabel?: string;
debugMeta?: Record<string, unknown>;
}
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<CanvasImageTaskQueueConfig> {
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<CanvasImageModelQueueStatus> {
@ -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<void>((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");
});

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

@ -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));
}

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

@ -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;
}

1
playwright.helpers.config.ts
View File

@ -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/,

37
tests/routes/canvas-image-queue-fairness.test.ts Normal file
View File

@ -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);
});
});