fix(batch): prevent status overwrite and stagger concurrent requests
Some checks failed
Lint / build (20) (push) Has been cancelled
Playwright Tests / test (push) Has been cancelled

This commit is contained in:
Your Name 2026-03-05 20:01:14 +08:00
parent 4a55cc1b0f
commit 4104b68f23
2 changed files with 115 additions and 41 deletions

View File

@ -37,6 +37,7 @@ const FALLBACK_PROMPT = "请参考第一张图片中的产品,将其自然地
const SCENE_FUSION_MODEL_COOLDOWN_MS = 2 * 60 * 1000;
const SCENE_FUSION_MAX_ITEMS = 10;
const SCENE_FUSION_REQUEST_STAGGER_MS = 350;
const sceneFusionModelCooldown = new Map<string, number>();
let sceneFusionRoundRobinCursor = 0;
@ -128,6 +129,10 @@ function getNextSceneFusionRoundRobinOffset(poolSize: number): number {
return offset;
}
function waitSceneFusion(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function buildSceneFusionCandidatePool(
allModels: Awaited<ReturnType<typeof getEnabledAiModelsWithProvider>>,
preferredModelId: string
@ -1064,6 +1069,11 @@ export async function POST(request: Request) {
nextSceneIndex += 1;
if (currentIndex >= processedScenes.length) break;
const staggerDelay = (currentIndex % workerCount) * SCENE_FUSION_REQUEST_STAGGER_MS;
if (staggerDelay > 0) {
await waitSceneFusion(staggerDelay);
}
const scene = processedScenes[currentIndex];
console.log(

View File

@ -26,6 +26,9 @@ import { ScrollArea } from "@/components/ui/scroll-area";
const fetcher = (url: string) => fetch(url).then((res) => res.json());
const MAX_BATCH_ITEMS = 10;
const BATCH_REQUEST_STAGGER_MS = 700;
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
type BatchMode = "scene-fission" | "multi-product-scene" | "random-bg" | "rmbg" | "enhance" | "watermark";
type ItemStatus = "pending" | "processing" | "completed" | "failed";
@ -415,29 +418,20 @@ export function SceneFusionView({ onBack, initialProductImage, initialProductNam
let buffer = "";
let success = 0;
let failed = 0;
let streamErrorMessage: string | null = null;
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) {
if (!line.trim()) continue;
const msg = JSON.parse(line) as {
type: string;
data?: {
id: string;
status: "completed" | "failed";
result?: string;
originalUrl?: string;
error?: string;
};
};
if (msg.type !== "result" || !msg.data) continue;
const handleParsedMessage = (msg: {
type: string;
data?: {
id: string;
status: "completed" | "failed";
result?: string;
originalUrl?: string;
error?: string;
};
error?: string;
}) => {
if (msg.type === "result" && msg.data) {
const data = msg.data;
if (data.status === "completed") {
success += 1;
@ -454,23 +448,88 @@ export function SceneFusionView({ onBack, initialProductImage, initialProductNam
: item
)
);
} else {
failed += 1;
setter((prev) =>
prev.map((item) =>
item.id === data.id
? {
...item,
status: "failed",
error: data.error || "生成失败",
}
: item
)
);
return;
}
failed += 1;
setter((prev) =>
prev.map((item) =>
item.id === data.id
? {
...item,
status: "failed",
error: data.error || "生成失败",
}
: item
)
);
return;
}
if (msg.type === "error") {
streamErrorMessage = msg.error || "处理失败";
}
};
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) {
if (!line.trim()) continue;
try {
const msg = JSON.parse(line) as {
type: string;
data?: {
id: string;
status: "completed" | "failed";
result?: string;
originalUrl?: string;
error?: string;
};
error?: string;
};
handleParsedMessage(msg);
} catch (parseError) {
console.warn("[SceneFusionView] Ignore invalid stream line:", parseError);
}
}
}
if (buffer.trim()) {
try {
const msg = JSON.parse(buffer.trim()) as {
type: string;
data?: {
id: string;
status: "completed" | "failed";
result?: string;
originalUrl?: string;
error?: string;
};
error?: string;
};
handleParsedMessage(msg);
} catch (parseError) {
console.warn("[SceneFusionView] Ignore trailing stream buffer:", parseError);
}
}
if (streamErrorMessage) {
setter((prev) =>
prev.map((item) =>
scenesPayload.some((scene) => scene.id === item.id) && item.status === "processing"
? { ...item, status: "failed", error: streamErrorMessage }
: item
)
);
toast.warning(streamErrorMessage);
}
if (success > 0 && failed === 0) {
toast.success(`批量处理完成:${success} 张`);
} else if (success > 0 || failed > 0) {
@ -479,14 +538,11 @@ export function SceneFusionView({ onBack, initialProductImage, initialProductNam
} catch (error) {
const message = error instanceof Error ? error.message : "处理失败";
toast.error(message);
updateFusionItemsStatus(
setter,
scenesPayload.map((item) => item.id),
"failed"
);
setter((prev) =>
prev.map((item) =>
scenesPayload.some((scene) => scene.id === item.id) ? { ...item, error: message } : item
scenesPayload.some((scene) => scene.id === item.id) && item.status === "processing"
? { ...item, status: "failed", error: message }
: item
)
);
} finally {
@ -655,6 +711,10 @@ export function SceneFusionView({ onBack, initialProductImage, initialProductNam
)
);
}
if (queue.length > 0) {
await sleep(BATCH_REQUEST_STAGGER_MS);
}
}
});
@ -804,6 +864,10 @@ export function SceneFusionView({ onBack, initialProductImage, initialProductNam
)
);
}
if (queue.length > 0) {
await sleep(BATCH_REQUEST_STAGGER_MS);
}
}
});