From 4104b68f238f2033046aae17e2b9c5bbbf252424 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 5 Mar 2026 20:01:14 +0800 Subject: [PATCH] fix(batch): prevent status overwrite and stagger concurrent requests --- app/api/scene-fusion/route.ts | 10 +++ components/scene-fusion-view.tsx | 146 ++++++++++++++++++++++--------- 2 files changed, 115 insertions(+), 41 deletions(-) diff --git a/app/api/scene-fusion/route.ts b/app/api/scene-fusion/route.ts index 5494b8da9..f76a79208 100644 --- a/app/api/scene-fusion/route.ts +++ b/app/api/scene-fusion/route.ts @@ -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(); let sceneFusionRoundRobinCursor = 0; @@ -128,6 +129,10 @@ function getNextSceneFusionRoundRobinOffset(poolSize: number): number { return offset; } +function waitSceneFusion(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + function buildSceneFusionCandidatePool( allModels: Awaited>, 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( diff --git a/components/scene-fusion-view.tsx b/components/scene-fusion-view.tsx index 9fd0eff61..ec86d202e 100644 --- a/components/scene-fusion-view.tsx +++ b/components/scene-fusion-view.tsx @@ -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); + } } });