fix: prevent import hanging by adding abort signals and idle timeouts
CI / check (push) Successful in 28s
CI / release (push) Skipped
CI / deploy (push) Successful in 59s

- Furni batch: wire request.signal to abort controller, send post_import
  progress events, skip post-import steps when aborted
- Clone sync-all: pre-fetch furnidata once per source instead of per item
  (eliminates 2000+ redundant DB reads + HTTP requests)
- SSE client: add 60s idle timeout to prevent infinite hangs when server
  stops responding
This commit is contained in:
openhands committed 2026-08-31 18:11:54 +02:00
1 parent 96c33efced
commit f70d96b81c
3 files changed
+119 -25

No files matched your search

@@ -6,6 +6,7 @@ import { logAudit } from "@/lib/services/audit";
import { import {
cloneSingleFurni, cloneSingleFurni,
fetchSourceFurnidata, fetchSourceFurnidata,
type SourceFurni,
} from "@/lib/services/clone-import"; } from "@/lib/services/clone-import";
import { listSources } from "@/lib/services/clone-sources"; import { listSources } from "@/lib/services/clone-sources";
import { import {
@@ -39,6 +40,20 @@ export const POST = withAdmin(
); );
} }
// Pre-fetch furnidata ONCE per source and build a lookup map.
// The old code re-fetched inside the worker loop for every item,
// causing 2000+ redundant DB reads + HTTP requests.
const sourceFurniDataMap = new Map<string, Map<string, SourceFurni>>();
for (const source of sources) {
try {
const entries = await fetchSourceFurnidata(source.furnidataUrl);
const byClassname = new Map(entries.map((e) => [e.classname, e]));
sourceFurniDataMap.set(source.id, byClassname);
} catch {}
}
// Collect all missing items using pre-fetched data
const allItems: Array<{ const allItems: Array<{
classname: string; classname: string;
sourceName: string; sourceName: string;
@@ -46,14 +61,10 @@ export const POST = withAdmin(
}> = []; }> = [];
for (const source of sources) { for (const source of sources) {
let entries: Awaited<ReturnType<typeof fetchSourceFurnidata>>; const byClassname = sourceFurniDataMap.get(source.id);
try { if (!byClassname) continue;
entries = await fetchSourceFurnidata(source.furnidataUrl);
} catch {
continue;
}
const classnames = entries.map((e) => e.classname); const classnames = [...byClassname.keys()];
if (classnames.length === 0) continue; if (classnames.length === 0) continue;
const rows = await db const rows = await db
@@ -62,10 +73,10 @@ export const POST = withAdmin(
.where(inArray(ItemsBase.itemName, classnames)); .where(inArray(ItemsBase.itemName, classnames));
const have = new Set(rows.map((r) => r.item_name)); const have = new Set(rows.map((r) => r.item_name));
for (const entry of entries) { for (const classname of classnames) {
if (!have.has(entry.classname)) { if (!have.has(classname)) {
allItems.push({ allItems.push({
classname: entry.classname, classname,
sourceName: source.name, sourceName: source.name,
sourceId: source.id, sourceId: source.id,
}); });
@@ -106,29 +117,30 @@ export const POST = withAdmin(
itemType: string; itemType: string;
}> = []; }> = [];
// Source lookup cache — pre-fetched above, no need to re-fetch per item
const sourceMap = new Map(sources.map((s) => [s.id, s]));
return runSseBatch({ return runSseBatch({
items: allItems, items: allItems,
concurrency: 10, concurrency: 10,
signal: request.signal, signal: request.signal,
labelOf: (it) => `${it.sourceName}/${it.classname}`, labelOf: (it) => `${it.sourceName}/${it.classname}`,
worker: async (it, _index, report) => { worker: async (it, _index, report) => {
const allSrc = await listSources(); const source = sourceMap.get(it.sourceId);
const source = allSrc.find((s) => s.id === it.sourceId);
if (!source) { if (!source) {
return { ok: false, error: "source not found" }; return { ok: false, error: "source not found" };
} }
let allEntries: Awaited<ReturnType<typeof fetchSourceFurnidata>>; // Use pre-fetched furnidata instead of re-fetching
try { const byClassname = sourceFurniDataMap.get(it.sourceId);
allEntries = await fetchSourceFurnidata(source.furnidataUrl); if (!byClassname) {
} catch (err) {
return { return {
ok: false, ok: false,
error: `furnidata fetch failed: ${(err as Error).message}`, error: "furnidata not available for this source",
}; };
} }
const entry = allEntries.find((e) => e.classname === it.classname); const entry = byClassname.get(it.classname);
if (!entry) { if (!entry) {
return { return {
ok: false, ok: false,
File diff suppressed because it is too large. Load diff
+24 -2
View File
@@ -3,26 +3,46 @@ import { adminFetch } from "@/lib/admin-fetch";
export type SseEvent = Record<string, unknown>; export type SseEvent = Record<string, unknown>;
/** Default idle timeout: if no SSE event arrives for this long, abort. */
const DEFAULT_IDLE_TIMEOUT_MS = 60_000;
/** /**
* Read an SSE response body and invoke `onEvent` for each `data:` JSON payload. * Read an SSE response body and invoke `onEvent` for each `data:` JSON payload.
* Aborts if no event arrives within `idleTimeoutMs` (default 60s).
*/ */
export async function readSseStream( export async function readSseStream(
body: ReadableStream<Uint8Array>, body: ReadableStream<Uint8Array>,
onEvent: (event: SseEvent) => void, onEvent: (event: SseEvent) => void,
signal?: AbortSignal, signal?: AbortSignal,
idleTimeoutMs: number = DEFAULT_IDLE_TIMEOUT_MS,
): Promise<void> { ): Promise<void> {
const reader = body.getReader(); const reader = body.getReader();
const decoder = new TextDecoder(); const decoder = new TextDecoder();
let buf = ""; let buf = "";
let idleTimer: ReturnType<typeof setTimeout> | undefined;
let done = false;
const resetIdleTimer = () => {
if (idleTimer) clearTimeout(idleTimer);
if (idleTimeoutMs > 0) {
idleTimer = setTimeout(() => {
if (!done) {
reader.cancel("SSE idle timeout").catch(() => {});
}
}, idleTimeoutMs);
}
};
try { try {
resetIdleTimer();
while (true) { while (true) {
if (signal?.aborted) { if (signal?.aborted) {
await reader.cancel(); await reader.cancel();
break; break;
} }
const { value, done } = await reader.read(); const { value, done: streamDone } = await reader.read();
if (done) break; if (streamDone) break;
resetIdleTimer();
buf += decoder.decode(value, { stream: true }); buf += decoder.decode(value, { stream: true });
const parts = buf.split("\n\n"); const parts = buf.split("\n\n");
buf = parts.pop() ?? ""; buf = parts.pop() ?? "";
@@ -36,6 +56,8 @@ export async function readSseStream(
} }
} }
} finally { } finally {
done = true;
if (idleTimer) clearTimeout(idleTimer);
reader.releaseLock(); reader.releaseLock();
} }
} }