mirror of
https://github.com/snapotter-hq/SnapOtter.git
synced 2026-08-03 07:46:42 +02:00
fix: honor unlimited processing timeout (#638)
* fix(web): recover stalled job progress streams * fix(ai): honor unlimited processing timeout * fix(web): keep retrying stalled progress streams
This commit is contained in:
@@ -90,7 +90,9 @@ export function useToolProcessor(toolId: string) {
|
|||||||
const abortRef = useRef<AbortController | null>(null);
|
const abortRef = useRef<AbortController | null>(null);
|
||||||
const stallTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
const stallTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
||||||
const activeJobIdRef = useRef<string | null>(null);
|
const activeJobIdRef = useRef<string | null>(null);
|
||||||
|
const activeEntryIndexRef = useRef<number | null>(null);
|
||||||
const asyncModeRef = useRef(false);
|
const asyncModeRef = useRef(false);
|
||||||
|
const reconnectSSERef = useRef<(force?: boolean) => void>(() => {});
|
||||||
// Save mode captured at run start (#495). Only "overwrite" re-anchors
|
// Save mode captured at run start (#495). Only "overwrite" re-anchors
|
||||||
// serverFileId to the saved result, so "new" keeps deriving from the
|
// serverFileId to the saved result, so "new" keeps deriving from the
|
||||||
// original library file on re-runs.
|
// original library file on re-runs.
|
||||||
@@ -101,6 +103,7 @@ export function useToolProcessor(toolId: string) {
|
|||||||
|
|
||||||
const clearActiveJob = useCallback(() => {
|
const clearActiveJob = useCallback(() => {
|
||||||
activeJobIdRef.current = null;
|
activeJobIdRef.current = null;
|
||||||
|
activeEntryIndexRef.current = null;
|
||||||
setActiveJob(null, null);
|
setActiveJob(null, null);
|
||||||
}, [setActiveJob]);
|
}, [setActiveJob]);
|
||||||
|
|
||||||
@@ -113,133 +116,134 @@ export function useToolProcessor(toolId: string) {
|
|||||||
headers: formatHeaders(),
|
headers: formatHeaders(),
|
||||||
});
|
});
|
||||||
} catch {
|
} catch {
|
||||||
// Cancel request failed; SSE handler or stall timeout will clean up
|
// Cancel request failed; SSE handler will clean up
|
||||||
}
|
}
|
||||||
}, []);
|
}, []);
|
||||||
|
|
||||||
const reconnectSSE = useCallback(() => {
|
const clearStallTimer = useCallback(() => {
|
||||||
const jobId = activeJobIdRef.current;
|
if (stallTimerRef.current) {
|
||||||
if (!jobId) return;
|
clearTimeout(stallTimerRef.current);
|
||||||
if (eventSourceRef.current && eventSourceRef.current.readyState === EventSource.OPEN) {
|
stallTimerRef.current = null;
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
|
}, []);
|
||||||
|
|
||||||
if (eventSourceRef.current) {
|
const resetStallTimer = useCallback(() => {
|
||||||
eventSourceRef.current.close();
|
clearStallTimer();
|
||||||
eventSourceRef.current = null;
|
stallTimerRef.current = setTimeout(() => {
|
||||||
}
|
stallTimerRef.current = null;
|
||||||
|
if (!activeJobIdRef.current || !asyncModeRef.current) return;
|
||||||
|
reconnectSSERef.current(true);
|
||||||
|
}, SSE_STALL_TIMEOUT_MS);
|
||||||
|
}, [clearStallTimer]);
|
||||||
|
|
||||||
try {
|
const reconnectSSE = useCallback(
|
||||||
const es = new EventSource(`/api/v1/jobs/${jobId}/progress`);
|
(force = false) => {
|
||||||
eventSourceRef.current = es;
|
const jobId = activeJobIdRef.current;
|
||||||
|
if (!jobId) return;
|
||||||
|
if (
|
||||||
|
!force &&
|
||||||
|
eventSourceRef.current &&
|
||||||
|
eventSourceRef.current.readyState === EventSource.OPEN
|
||||||
|
) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
es.onmessage = (event) => {
|
if (eventSourceRef.current) {
|
||||||
try {
|
eventSourceRef.current.close();
|
||||||
const data = JSON.parse(event.data);
|
eventSourceRef.current = null;
|
||||||
if (data.type === "heartbeat") {
|
}
|
||||||
if (asyncModeRef.current && stallTimerRef.current) {
|
|
||||||
clearTimeout(stallTimerRef.current);
|
try {
|
||||||
stallTimerRef.current = setTimeout(() => {
|
const es = new EventSource(`/api/v1/jobs/${jobId}/progress`);
|
||||||
if (eventSourceRef.current) {
|
eventSourceRef.current = es;
|
||||||
eventSourceRef.current.close();
|
if (asyncModeRef.current) resetStallTimer();
|
||||||
eventSourceRef.current = null;
|
|
||||||
}
|
es.onmessage = (event) => {
|
||||||
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
if (eventSourceRef.current !== es) return;
|
||||||
clearActiveJob();
|
try {
|
||||||
setError(
|
const data = JSON.parse(event.data);
|
||||||
"Processing timed out after 5 minutes without an update from the server. Heavy tools run much slower on CPU than a GPU; try a smaller file, or retry if the connection dropped.",
|
if (data.type === "heartbeat") {
|
||||||
);
|
if (asyncModeRef.current) resetStallTimer();
|
||||||
setProcessing(false);
|
return;
|
||||||
setProgress(IDLE_PROGRESS);
|
|
||||||
}, SSE_STALL_TIMEOUT_MS);
|
|
||||||
}
|
}
|
||||||
return;
|
if (data.type !== "single") return;
|
||||||
}
|
|
||||||
if (data.type !== "single") return;
|
|
||||||
|
|
||||||
if (asyncModeRef.current && stallTimerRef.current) {
|
if (asyncModeRef.current) resetStallTimer();
|
||||||
clearTimeout(stallTimerRef.current);
|
|
||||||
stallTimerRef.current = setTimeout(() => {
|
if (data.phase === "complete" && data.result) {
|
||||||
if (eventSourceRef.current) {
|
clearStallTimer();
|
||||||
eventSourceRef.current.close();
|
|
||||||
eventSourceRef.current = null;
|
|
||||||
}
|
|
||||||
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
||||||
|
es.close();
|
||||||
|
eventSourceRef.current = null;
|
||||||
|
const idx = activeEntryIndexRef.current ?? useFileStore.getState().selectedIndex;
|
||||||
|
|
||||||
|
const result = data.result as ProcessResult;
|
||||||
|
setWarning(result.warning ?? null);
|
||||||
|
setResultPayload(result as unknown as Record<string, unknown>);
|
||||||
|
if (result.savedFileId) {
|
||||||
|
useFileStore.getState().setLastSavedLibraryFileId(result.savedFileId);
|
||||||
|
}
|
||||||
|
useFileStore.getState().updateEntry(idx, {
|
||||||
|
processedUrl: result.downloadUrl,
|
||||||
|
processedPreviewUrl: result.previewUrl ?? null,
|
||||||
|
processedFilename: null,
|
||||||
|
status: "completed",
|
||||||
|
originalSize: result.originalSize,
|
||||||
|
processedSize: result.processedSize,
|
||||||
|
...(result.savedFileId && saveModeRef.current === "overwrite"
|
||||||
|
? { serverFileId: result.savedFileId }
|
||||||
|
: {}),
|
||||||
|
});
|
||||||
clearActiveJob();
|
clearActiveJob();
|
||||||
setError(
|
|
||||||
"Processing timed out after 5 minutes without an update from the server. Heavy tools run much slower on CPU than a GPU; try a smaller file, or retry if the connection dropped.",
|
|
||||||
);
|
|
||||||
setProcessing(false);
|
setProcessing(false);
|
||||||
setProgress(IDLE_PROGRESS);
|
setProgress(IDLE_PROGRESS);
|
||||||
}, SSE_STALL_TIMEOUT_MS);
|
return;
|
||||||
}
|
|
||||||
|
|
||||||
if (data.phase === "complete" && data.result) {
|
|
||||||
if (stallTimerRef.current) clearTimeout(stallTimerRef.current);
|
|
||||||
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
|
||||||
es.close();
|
|
||||||
eventSourceRef.current = null;
|
|
||||||
clearActiveJob();
|
|
||||||
|
|
||||||
const result = data.result as ProcessResult;
|
|
||||||
setWarning(result.warning ?? null);
|
|
||||||
setResultPayload(result as unknown as Record<string, unknown>);
|
|
||||||
if (result.savedFileId) {
|
|
||||||
useFileStore.getState().setLastSavedLibraryFileId(result.savedFileId);
|
|
||||||
}
|
}
|
||||||
const idx = useFileStore.getState().selectedIndex;
|
|
||||||
useFileStore.getState().updateEntry(idx, {
|
|
||||||
processedUrl: result.downloadUrl,
|
|
||||||
processedPreviewUrl: result.previewUrl ?? null,
|
|
||||||
processedFilename: null,
|
|
||||||
status: "completed",
|
|
||||||
originalSize: result.originalSize,
|
|
||||||
processedSize: result.processedSize,
|
|
||||||
...(result.savedFileId && saveModeRef.current === "overwrite"
|
|
||||||
? { serverFileId: result.savedFileId }
|
|
||||||
: {}),
|
|
||||||
});
|
|
||||||
setProcessing(false);
|
|
||||||
setProgress(IDLE_PROGRESS);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (data.phase === "failed") {
|
if (data.phase === "failed") {
|
||||||
if (stallTimerRef.current) clearTimeout(stallTimerRef.current);
|
clearStallTimer();
|
||||||
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
||||||
|
es.close();
|
||||||
|
eventSourceRef.current = null;
|
||||||
|
clearActiveJob();
|
||||||
|
setError(data.error || "Processing failed");
|
||||||
|
setProcessing(false);
|
||||||
|
setProgress(IDLE_PROGRESS);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (typeof data.percent === "number") {
|
||||||
|
const scaled = UPLOAD_WEIGHT + (data.percent / 100) * (100 - UPLOAD_WEIGHT);
|
||||||
|
setProgress((prev) => ({
|
||||||
|
...prev,
|
||||||
|
phase: "processing",
|
||||||
|
percent: Math.max(prev.percent, scaled),
|
||||||
|
stage: data.stage,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
} catch {
|
||||||
|
// Ignore malformed SSE
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
es.onerror = () => {
|
||||||
|
if (!asyncModeRef.current) {
|
||||||
es.close();
|
es.close();
|
||||||
eventSourceRef.current = null;
|
if (eventSourceRef.current === es) {
|
||||||
clearActiveJob();
|
eventSourceRef.current = null;
|
||||||
setError(data.error || "Processing failed");
|
}
|
||||||
setProcessing(false);
|
|
||||||
setProgress(IDLE_PROGRESS);
|
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
|
};
|
||||||
|
} catch {
|
||||||
|
// EventSource creation failed
|
||||||
|
}
|
||||||
|
},
|
||||||
|
[clearActiveJob, clearStallTimer, resetStallTimer, setError, setProcessing],
|
||||||
|
);
|
||||||
|
|
||||||
if (typeof data.percent === "number") {
|
useEffect(() => {
|
||||||
const scaled = UPLOAD_WEIGHT + (data.percent / 100) * (100 - UPLOAD_WEIGHT);
|
reconnectSSERef.current = reconnectSSE;
|
||||||
setProgress((prev) => ({
|
}, [reconnectSSE]);
|
||||||
...prev,
|
|
||||||
phase: "processing",
|
|
||||||
percent: Math.max(prev.percent, scaled),
|
|
||||||
stage: data.stage,
|
|
||||||
}));
|
|
||||||
}
|
|
||||||
} catch {
|
|
||||||
// Ignore malformed SSE
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
es.onerror = () => {
|
|
||||||
if (!asyncModeRef.current) {
|
|
||||||
es.close();
|
|
||||||
eventSourceRef.current = null;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
} catch {
|
|
||||||
// EventSource creation failed
|
|
||||||
}
|
|
||||||
}, [clearActiveJob, setError, setProcessing]);
|
|
||||||
|
|
||||||
// Reconnect SSE when tab becomes visible again (mobile tab recovery)
|
// Reconnect SSE when tab becomes visible again (mobile tab recovery)
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
@@ -305,118 +309,11 @@ export function useToolProcessor(toolId: string) {
|
|||||||
|
|
||||||
const clientJobId = generateId();
|
const clientJobId = generateId();
|
||||||
activeJobIdRef.current = clientJobId;
|
activeJobIdRef.current = clientJobId;
|
||||||
|
activeEntryIndexRef.current = capturedIndex;
|
||||||
asyncModeRef.current = false;
|
asyncModeRef.current = false;
|
||||||
let asyncMode = false;
|
|
||||||
|
|
||||||
const clearStallTimer = () => {
|
|
||||||
if (stallTimerRef.current) {
|
|
||||||
clearTimeout(stallTimerRef.current);
|
|
||||||
stallTimerRef.current = null;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
const resetStallTimer = () => {
|
|
||||||
clearStallTimer();
|
|
||||||
stallTimerRef.current = setTimeout(() => {
|
|
||||||
if (eventSourceRef.current) {
|
|
||||||
eventSourceRef.current.close();
|
|
||||||
eventSourceRef.current = null;
|
|
||||||
}
|
|
||||||
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
|
||||||
clearActiveJob();
|
|
||||||
useFileStore.getState().updateEntry(capturedIndex, {
|
|
||||||
status: "failed",
|
|
||||||
error: "Processing timed out",
|
|
||||||
});
|
|
||||||
setError(
|
|
||||||
"Processing timed out after 5 minutes without an update from the server. Heavy tools run much slower on CPU than a GPU; try a smaller file, or retry if the connection dropped.",
|
|
||||||
);
|
|
||||||
setProcessing(false);
|
|
||||||
setProgress(IDLE_PROGRESS);
|
|
||||||
}, SSE_STALL_TIMEOUT_MS);
|
|
||||||
};
|
|
||||||
|
|
||||||
// Open SSE for real-time progress from the server (all tools)
|
// Open SSE for real-time progress from the server (all tools)
|
||||||
try {
|
reconnectSSE(true);
|
||||||
const es = new EventSource(`/api/v1/jobs/${clientJobId}/progress`);
|
|
||||||
eventSourceRef.current = es;
|
|
||||||
|
|
||||||
es.onmessage = (event) => {
|
|
||||||
try {
|
|
||||||
const data = JSON.parse(event.data);
|
|
||||||
if (data.type === "heartbeat") {
|
|
||||||
if (asyncMode) resetStallTimer();
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
if (data.type !== "single") return;
|
|
||||||
|
|
||||||
if (asyncMode) resetStallTimer();
|
|
||||||
|
|
||||||
// AI tools deliver results via SSE (they return 202 from the XHR)
|
|
||||||
if (data.phase === "complete" && data.result) {
|
|
||||||
clearStallTimer();
|
|
||||||
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
|
||||||
es.close();
|
|
||||||
eventSourceRef.current = null;
|
|
||||||
clearActiveJob();
|
|
||||||
|
|
||||||
const result = data.result as ProcessResult;
|
|
||||||
setWarning(result.warning ?? null);
|
|
||||||
setResultPayload(result as unknown as Record<string, unknown>);
|
|
||||||
if (result.savedFileId) {
|
|
||||||
useFileStore.getState().setLastSavedLibraryFileId(result.savedFileId);
|
|
||||||
}
|
|
||||||
useFileStore.getState().updateEntry(capturedIndex, {
|
|
||||||
processedUrl: result.downloadUrl,
|
|
||||||
processedPreviewUrl: result.previewUrl ?? null,
|
|
||||||
processedFilename: null,
|
|
||||||
status: "completed",
|
|
||||||
originalSize: result.originalSize,
|
|
||||||
processedSize: result.processedSize,
|
|
||||||
...(result.savedFileId && saveModeRef.current === "overwrite"
|
|
||||||
? { serverFileId: result.savedFileId }
|
|
||||||
: {}),
|
|
||||||
});
|
|
||||||
setProcessing(false);
|
|
||||||
setProgress(IDLE_PROGRESS);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (data.phase === "failed" && asyncMode) {
|
|
||||||
clearStallTimer();
|
|
||||||
if (elapsedRef.current) clearInterval(elapsedRef.current);
|
|
||||||
es.close();
|
|
||||||
eventSourceRef.current = null;
|
|
||||||
clearActiveJob();
|
|
||||||
setError(data.error || "Processing failed");
|
|
||||||
setProcessing(false);
|
|
||||||
setProgress(IDLE_PROGRESS);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (typeof data.percent === "number") {
|
|
||||||
const scaled = UPLOAD_WEIGHT + (data.percent / 100) * (100 - UPLOAD_WEIGHT);
|
|
||||||
setProgress((prev) => ({
|
|
||||||
...prev,
|
|
||||||
phase: "processing",
|
|
||||||
percent: Math.max(prev.percent, scaled),
|
|
||||||
stage: data.stage,
|
|
||||||
}));
|
|
||||||
}
|
|
||||||
} catch {
|
|
||||||
// Ignore malformed SSE
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
es.onerror = () => {
|
|
||||||
if (!asyncMode) {
|
|
||||||
es.close();
|
|
||||||
eventSourceRef.current = null;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
} catch {
|
|
||||||
// EventSource creation failed -- proceed without SSE
|
|
||||||
}
|
|
||||||
|
|
||||||
// Build form data
|
// Build form data
|
||||||
const cleanSettings = { ...settings };
|
const cleanSettings = { ...settings };
|
||||||
@@ -471,7 +368,6 @@ export function useToolProcessor(toolId: string) {
|
|||||||
|
|
||||||
xhr.onload = () => {
|
xhr.onload = () => {
|
||||||
if (xhr.status === 202) {
|
if (xhr.status === 202) {
|
||||||
asyncMode = true;
|
|
||||||
asyncModeRef.current = true;
|
asyncModeRef.current = true;
|
||||||
setActiveJob(clientJobId, cancelCurrentJob);
|
setActiveJob(clientJobId, cancelCurrentJob);
|
||||||
resetStallTimer();
|
resetStallTimer();
|
||||||
@@ -567,6 +463,9 @@ export function useToolProcessor(toolId: string) {
|
|||||||
setActiveJob,
|
setActiveJob,
|
||||||
clearActiveJob,
|
clearActiveJob,
|
||||||
cancelCurrentJob,
|
cancelCurrentJob,
|
||||||
|
clearStallTimer,
|
||||||
|
reconnectSSE,
|
||||||
|
resetStallTimer,
|
||||||
toolName,
|
toolName,
|
||||||
],
|
],
|
||||||
);
|
);
|
||||||
|
|||||||
+44
-30
@@ -9,12 +9,25 @@ import { acquireVenvRead, tryAcquireVenvRead } from "./venv-lock.js";
|
|||||||
|
|
||||||
const __dirname = dirname(fileURLToPath(import.meta.url));
|
const __dirname = dirname(fileURLToPath(import.meta.url));
|
||||||
const PYTHON_DIR = resolve(__dirname, "../python");
|
const PYTHON_DIR = resolve(__dirname, "../python");
|
||||||
|
const DEFAULT_PYTHON_TIMEOUT_MS = 600_000;
|
||||||
|
|
||||||
function appendEnvPath(base: string, suffix: string): string {
|
function appendEnvPath(base: string, suffix: string): string {
|
||||||
const normalizedBase = base.replace(/\/+$/, "");
|
const normalizedBase = base.replace(/\/+$/, "");
|
||||||
return `${normalizedBase || "/"}${normalizedBase === "" ? "" : "/"}${suffix}`;
|
return `${normalizedBase || "/"}${normalizedBase === "" ? "" : "/"}${suffix}`;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function resolvePythonTimeout(explicitTimeout?: number): number | undefined {
|
||||||
|
const configured = process.env.PROCESSING_TIMEOUT_S?.trim();
|
||||||
|
if (configured !== undefined && configured !== "") {
|
||||||
|
const seconds = Number(configured);
|
||||||
|
if (Number.isFinite(seconds)) {
|
||||||
|
if (seconds === 0) return undefined;
|
||||||
|
if (seconds > 0) return seconds * 1000;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return explicitTimeout ?? DEFAULT_PYTHON_TIMEOUT_MS;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Build a minimal environment for spawned Python processes.
|
* Build a minimal environment for spawned Python processes.
|
||||||
* Only passes through variables needed for venv, CUDA, model cache,
|
* Only passes through variables needed for venv, CUDA, model cache,
|
||||||
@@ -476,32 +489,34 @@ export class PythonDispatcher {
|
|||||||
if (!proc || !proc.stdin || !this.childReady) return null;
|
if (!proc || !proc.stdin || !this.childReady) return null;
|
||||||
|
|
||||||
const id = randomUUID();
|
const id = randomUUID();
|
||||||
const timeout =
|
const timeout = resolvePythonTimeout(options.timeout);
|
||||||
options.timeout ??
|
|
||||||
(process.env.PROCESSING_TIMEOUT_S && parseInt(process.env.PROCESSING_TIMEOUT_S, 10) > 0
|
|
||||||
? parseInt(process.env.PROCESSING_TIMEOUT_S, 10) * 1000
|
|
||||||
: 600000);
|
|
||||||
|
|
||||||
return new Promise((resolvePromise, rejectPromise) => {
|
return new Promise((resolvePromise, rejectPromise) => {
|
||||||
const timer = setTimeout(() => {
|
const timer =
|
||||||
this.pending.delete(id);
|
timeout === undefined
|
||||||
// Kill the stuck dispatcher so it restarts on the next request instead of
|
? undefined
|
||||||
// blocking all subsequent operations behind the timed-out script.
|
: setTimeout(() => {
|
||||||
if (this.child && !this.child.killed) {
|
this.pending.delete(id);
|
||||||
this.child.kill("SIGTERM");
|
// Kill the stuck dispatcher so it restarts on the next request instead of
|
||||||
}
|
// blocking all subsequent operations behind the timed-out script.
|
||||||
rejectPromise(
|
if (this.child && !this.child.killed) {
|
||||||
new SafeError("Python script timed out", { kind: "operational", code: "timeout" }),
|
this.child.kill("SIGTERM");
|
||||||
);
|
}
|
||||||
}, timeout);
|
rejectPromise(
|
||||||
|
new SafeError("Python script timed out", {
|
||||||
|
kind: "operational",
|
||||||
|
code: "timeout",
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
}, timeout);
|
||||||
|
|
||||||
const wrappedResolve = (result: { stdout: string; stderr: string }) => {
|
const wrappedResolve = (result: { stdout: string; stderr: string }) => {
|
||||||
clearTimeout(timer);
|
if (timer) clearTimeout(timer);
|
||||||
resolvePromise(result);
|
resolvePromise(result);
|
||||||
};
|
};
|
||||||
|
|
||||||
const wrappedReject = (err: Error) => {
|
const wrappedReject = (err: Error) => {
|
||||||
clearTimeout(timer);
|
if (timer) clearTimeout(timer);
|
||||||
rejectPromise(err);
|
rejectPromise(err);
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -529,7 +544,7 @@ export class PythonDispatcher {
|
|||||||
proc.stdin!.write(request + "\n");
|
proc.stdin!.write(request + "\n");
|
||||||
} catch {
|
} catch {
|
||||||
this.pending.delete(id);
|
this.pending.delete(id);
|
||||||
clearTimeout(timer);
|
if (timer) clearTimeout(timer);
|
||||||
rejectPromise(
|
rejectPromise(
|
||||||
new SafeError("Python dispatcher stdin closed unexpectedly", {
|
new SafeError("Python dispatcher stdin closed unexpectedly", {
|
||||||
kind: "operational",
|
kind: "operational",
|
||||||
@@ -564,11 +579,7 @@ export class PythonDispatcher {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
const scriptPath = resolve(PYTHON_DIR, scriptName);
|
const scriptPath = resolve(PYTHON_DIR, scriptName);
|
||||||
const timeout =
|
const timeout = resolvePythonTimeout(options.timeout);
|
||||||
options.timeout ??
|
|
||||||
(process.env.PROCESSING_TIMEOUT_S && parseInt(process.env.PROCESSING_TIMEOUT_S, 10) > 0
|
|
||||||
? parseInt(process.env.PROCESSING_TIMEOUT_S, 10) * 1000
|
|
||||||
: 600000);
|
|
||||||
|
|
||||||
return new Promise((resolvePromise, rejectPromise) => {
|
return new Promise((resolvePromise, rejectPromise) => {
|
||||||
const trySpawn = (pythonBin: string, isFallback: boolean) => {
|
const trySpawn = (pythonBin: string, isFallback: boolean) => {
|
||||||
@@ -582,10 +593,13 @@ export class PythonDispatcher {
|
|||||||
let stderrBuffer = "";
|
let stderrBuffer = "";
|
||||||
let timedOut = false;
|
let timedOut = false;
|
||||||
|
|
||||||
const timer = setTimeout(() => {
|
const timer =
|
||||||
timedOut = true;
|
timeout === undefined
|
||||||
proc.kill("SIGTERM");
|
? undefined
|
||||||
}, timeout);
|
: setTimeout(() => {
|
||||||
|
timedOut = true;
|
||||||
|
proc.kill("SIGTERM");
|
||||||
|
}, timeout);
|
||||||
|
|
||||||
proc.stdout.on("data", (chunk: Buffer) => {
|
proc.stdout.on("data", (chunk: Buffer) => {
|
||||||
stdout += chunk.toString();
|
stdout += chunk.toString();
|
||||||
@@ -614,7 +628,7 @@ export class PythonDispatcher {
|
|||||||
});
|
});
|
||||||
|
|
||||||
proc.on("error", (err: NodeJS.ErrnoException) => {
|
proc.on("error", (err: NodeJS.ErrnoException) => {
|
||||||
clearTimeout(timer);
|
if (timer) clearTimeout(timer);
|
||||||
if (err.code === "ENOENT" && !isFallback) {
|
if (err.code === "ENOENT" && !isFallback) {
|
||||||
trySpawn("python3", true);
|
trySpawn("python3", true);
|
||||||
} else {
|
} else {
|
||||||
@@ -628,7 +642,7 @@ export class PythonDispatcher {
|
|||||||
});
|
});
|
||||||
|
|
||||||
proc.on("close", (code, signal) => {
|
proc.on("close", (code, signal) => {
|
||||||
clearTimeout(timer);
|
if (timer) clearTimeout(timer);
|
||||||
|
|
||||||
if (stderrBuffer.trim()) {
|
if (stderrBuffer.trim()) {
|
||||||
stderrLines.push(stderrBuffer.trim());
|
stderrLines.push(stderrBuffer.trim());
|
||||||
|
|||||||
@@ -712,7 +712,7 @@ describe("bridge - PROCESSING_TIMEOUT_S env for dispatcher path", () => {
|
|||||||
vi.useRealTimers();
|
vi.useRealTimers();
|
||||||
});
|
});
|
||||||
|
|
||||||
it("ignores zero PROCESSING_TIMEOUT_S and uses default 600s", async () => {
|
it("treats zero PROCESSING_TIMEOUT_S as unlimited despite an explicit timeout", async () => {
|
||||||
process.env.PROCESSING_TIMEOUT_S = "0";
|
process.env.PROCESSING_TIMEOUT_S = "0";
|
||||||
|
|
||||||
vi.useFakeTimers();
|
vi.useFakeTimers();
|
||||||
@@ -724,13 +724,16 @@ describe("bridge - PROCESSING_TIMEOUT_S env for dispatcher path", () => {
|
|||||||
vi.advanceTimersByTime(100);
|
vi.advanceTimersByTime(100);
|
||||||
await initPromise;
|
await initPromise;
|
||||||
|
|
||||||
const promise = runPythonWithProgress("test.py", []);
|
const promise = runPythonWithProgress("test.py", [], { timeout: 50 });
|
||||||
|
const settled = promise.then(
|
||||||
|
(result) => ({ status: "resolved" as const, result }),
|
||||||
|
(error: Error) => ({ status: "rejected" as const, error }),
|
||||||
|
);
|
||||||
await vi.advanceTimersByTimeAsync(10);
|
await vi.advanceTimersByTimeAsync(10);
|
||||||
|
|
||||||
// Advance 10s -- should NOT have timed out with default 600s
|
vi.advanceTimersByTime(600_001);
|
||||||
vi.advanceTimersByTime(10_000);
|
expect(mock.process.kill).not.toHaveBeenCalled();
|
||||||
|
|
||||||
// Respond before default timeout
|
|
||||||
const line = mock.stdinWrites.join("").split("\n").filter(Boolean)[0];
|
const line = mock.stdinWrites.join("").split("\n").filter(Boolean)[0];
|
||||||
const id = JSON.parse(line).id;
|
const id = JSON.parse(line).id;
|
||||||
mock.stdout.emit(
|
mock.stdout.emit(
|
||||||
@@ -738,8 +741,33 @@ describe("bridge - PROCESSING_TIMEOUT_S env for dispatcher path", () => {
|
|||||||
Buffer.from(`${JSON.stringify({ id, exitCode: 0, stdout: '{"ok":true}' })}\n`),
|
Buffer.from(`${JSON.stringify({ id, exitCode: 0, stdout: '{"ok":true}' })}\n`),
|
||||||
);
|
);
|
||||||
|
|
||||||
const result = await promise;
|
const outcome = await settled;
|
||||||
expect(result.stdout).toBe('{"ok":true}');
|
expect(outcome).toMatchObject({
|
||||||
|
status: "resolved",
|
||||||
|
result: { stdout: '{"ok":true}' },
|
||||||
|
});
|
||||||
|
vi.useRealTimers();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("lets positive PROCESSING_TIMEOUT_S override an explicit timeout", async () => {
|
||||||
|
process.env.PROCESSING_TIMEOUT_S = "2";
|
||||||
|
|
||||||
|
vi.useFakeTimers();
|
||||||
|
const mock = createMockProcess();
|
||||||
|
vi.mocked(spawn).mockReturnValue(mock.process);
|
||||||
|
|
||||||
|
const initPromise = initDispatcher();
|
||||||
|
mock.stderr.emit("data", Buffer.from('{"ready": true, "gpu": false}\n'));
|
||||||
|
vi.advanceTimersByTime(100);
|
||||||
|
await initPromise;
|
||||||
|
|
||||||
|
const promise = runPythonWithProgress("test.py", [], { timeout: 60_000 });
|
||||||
|
const rejection = expect(promise).rejects.toThrow("Python script timed out");
|
||||||
|
await vi.advanceTimersByTimeAsync(10);
|
||||||
|
|
||||||
|
vi.advanceTimersByTime(2_500);
|
||||||
|
|
||||||
|
await rejection;
|
||||||
vi.useRealTimers();
|
vi.useRealTimers();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -619,25 +619,57 @@ describe("bridge - runPythonWithProgress (per-request fallback)", () => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
it("ignores invalid PROCESSING_TIMEOUT_S values", async () => {
|
it("treats zero PROCESSING_TIMEOUT_S as unlimited despite an explicit timeout", async () => {
|
||||||
const origTimeout = process.env.PROCESSING_TIMEOUT_S;
|
const origTimeout = process.env.PROCESSING_TIMEOUT_S;
|
||||||
process.env.PROCESSING_TIMEOUT_S = "0";
|
process.env.PROCESSING_TIMEOUT_S = "0";
|
||||||
|
|
||||||
const mock = createMockProcess();
|
vi.useFakeTimers();
|
||||||
vi.mocked(spawn).mockReturnValue(mock.process);
|
try {
|
||||||
|
const mock = createMockProcess();
|
||||||
|
vi.mocked(spawn).mockReturnValue(mock.process);
|
||||||
|
|
||||||
const promise = runPythonWithProgress("test_script.py", []);
|
const promise = runPythonWithProgress("test_script.py", [], { timeout: 50 });
|
||||||
|
|
||||||
mock.stdout.emit("data", Buffer.from('{"success": true}\n'));
|
vi.advanceTimersByTime(600_001);
|
||||||
mock.emitEvent("close", 0, null);
|
expect(mock.process.kill).not.toHaveBeenCalled();
|
||||||
|
|
||||||
// Should not throw -- falls back to 600000ms default
|
mock.stdout.emit("data", Buffer.from('{"success": true}\n'));
|
||||||
await expect(promise).resolves.toBeDefined();
|
mock.emitEvent("close", 0, null);
|
||||||
|
|
||||||
if (origTimeout !== undefined) {
|
await expect(promise).resolves.toBeDefined();
|
||||||
process.env.PROCESSING_TIMEOUT_S = origTimeout;
|
} finally {
|
||||||
} else {
|
vi.useRealTimers();
|
||||||
delete process.env.PROCESSING_TIMEOUT_S;
|
if (origTimeout !== undefined) {
|
||||||
|
process.env.PROCESSING_TIMEOUT_S = origTimeout;
|
||||||
|
} else {
|
||||||
|
delete process.env.PROCESSING_TIMEOUT_S;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("lets positive PROCESSING_TIMEOUT_S override an explicit timeout", async () => {
|
||||||
|
const origTimeout = process.env.PROCESSING_TIMEOUT_S;
|
||||||
|
process.env.PROCESSING_TIMEOUT_S = "2";
|
||||||
|
|
||||||
|
vi.useFakeTimers();
|
||||||
|
try {
|
||||||
|
const mock = createMockProcess();
|
||||||
|
vi.mocked(spawn).mockReturnValue(mock.process);
|
||||||
|
|
||||||
|
const promise = runPythonWithProgress("test_script.py", [], { timeout: 60_000 });
|
||||||
|
const rejection = expect(promise).rejects.toThrow("Python script timed out");
|
||||||
|
|
||||||
|
vi.advanceTimersByTime(2_500);
|
||||||
|
mock.emitEvent("close", null, "SIGTERM");
|
||||||
|
|
||||||
|
await rejection;
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers();
|
||||||
|
if (origTimeout !== undefined) {
|
||||||
|
process.env.PROCESSING_TIMEOUT_S = origTimeout;
|
||||||
|
} else {
|
||||||
|
delete process.env.PROCESSING_TIMEOUT_S;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -0,0 +1,158 @@
|
|||||||
|
// @vitest-environment jsdom
|
||||||
|
|
||||||
|
import { act, renderHook } from "@testing-library/react";
|
||||||
|
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||||
|
|
||||||
|
vi.mock("@/lib/image-preview", () => ({
|
||||||
|
needsServerPreview: vi.fn(() => false),
|
||||||
|
fetchDecodedPreview: vi.fn(() => Promise.resolve(null)),
|
||||||
|
}));
|
||||||
|
|
||||||
|
vi.mock("@/lib/analytics", () => ({
|
||||||
|
track: vi.fn(),
|
||||||
|
}));
|
||||||
|
|
||||||
|
vi.mock("@/lib/api", () => ({
|
||||||
|
formatHeaders: () => new Map<string, string>(),
|
||||||
|
parseApiError: () => "error",
|
||||||
|
}));
|
||||||
|
|
||||||
|
vi.mock("@/lib/utils", async (importOriginal) => {
|
||||||
|
const actual: Record<string, unknown> = await importOriginal();
|
||||||
|
return { ...actual, generateId: () => "11111111-1111-4111-8111-111111111111" };
|
||||||
|
});
|
||||||
|
|
||||||
|
import { useToolProcessor } from "@/hooks/use-tool-processor";
|
||||||
|
import { useFileStore } from "@/stores/file-store";
|
||||||
|
|
||||||
|
interface MockXhr {
|
||||||
|
status: number;
|
||||||
|
responseText: string;
|
||||||
|
timeout: number;
|
||||||
|
upload: { onprogress?: unknown; onload?: unknown };
|
||||||
|
onload?: () => void;
|
||||||
|
onerror?: (() => void) | null;
|
||||||
|
ontimeout?: (() => void) | null;
|
||||||
|
open: ReturnType<typeof vi.fn>;
|
||||||
|
send: ReturnType<typeof vi.fn>;
|
||||||
|
setRequestHeader: ReturnType<typeof vi.fn>;
|
||||||
|
abort: ReturnType<typeof vi.fn>;
|
||||||
|
}
|
||||||
|
|
||||||
|
class MockEventSource {
|
||||||
|
static OPEN = 1;
|
||||||
|
static instances: MockEventSource[] = [];
|
||||||
|
|
||||||
|
onmessage: ((event: MessageEvent) => void) | null = null;
|
||||||
|
onerror: (() => void) | null = null;
|
||||||
|
readyState = MockEventSource.OPEN;
|
||||||
|
close = vi.fn(() => {
|
||||||
|
this.readyState = 2;
|
||||||
|
});
|
||||||
|
|
||||||
|
constructor(readonly url: string) {
|
||||||
|
MockEventSource.instances.push(this);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let xhrs: MockXhr[];
|
||||||
|
|
||||||
|
beforeEach(() => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
vi.stubGlobal("URL", {
|
||||||
|
...globalThis.URL,
|
||||||
|
createObjectURL: vi.fn(() => "blob:fake-url"),
|
||||||
|
revokeObjectURL: vi.fn(),
|
||||||
|
});
|
||||||
|
useFileStore.getState().reset();
|
||||||
|
xhrs = [];
|
||||||
|
MockEventSource.instances = [];
|
||||||
|
vi.stubGlobal("EventSource", MockEventSource);
|
||||||
|
vi.stubGlobal(
|
||||||
|
"XMLHttpRequest",
|
||||||
|
vi.fn(() => {
|
||||||
|
const xhr: MockXhr = {
|
||||||
|
status: 0,
|
||||||
|
responseText: "",
|
||||||
|
timeout: 0,
|
||||||
|
upload: {},
|
||||||
|
open: vi.fn(),
|
||||||
|
send: vi.fn(),
|
||||||
|
setRequestHeader: vi.fn(),
|
||||||
|
abort: vi.fn(),
|
||||||
|
};
|
||||||
|
xhrs.push(xhr);
|
||||||
|
return xhr;
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
afterEach(() => {
|
||||||
|
vi.unstubAllGlobals();
|
||||||
|
vi.useRealTimers();
|
||||||
|
});
|
||||||
|
|
||||||
|
describe("useToolProcessor SSE recovery", () => {
|
||||||
|
it("reconnects after a transport stall and accepts the terminal replay", () => {
|
||||||
|
const file = new File([new ArrayBuffer(64)], "photo.png", { type: "image/png" });
|
||||||
|
useFileStore.getState().setFiles([file]);
|
||||||
|
const { result, unmount } = renderHook(() => useToolProcessor("upscale"));
|
||||||
|
|
||||||
|
act(() => {
|
||||||
|
result.current.processFiles([file], {});
|
||||||
|
});
|
||||||
|
act(() => {
|
||||||
|
xhrs[0].status = 202;
|
||||||
|
xhrs[0].responseText = JSON.stringify({ jobId: "server-job", async: true });
|
||||||
|
xhrs[0].onload?.();
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(MockEventSource.instances).toHaveLength(1);
|
||||||
|
expect(useFileStore.getState().activeJobId).toBe("11111111-1111-4111-8111-111111111111");
|
||||||
|
|
||||||
|
act(() => {
|
||||||
|
vi.advanceTimersByTime(300_001);
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(MockEventSource.instances[0].close).toHaveBeenCalledOnce();
|
||||||
|
expect(MockEventSource.instances).toHaveLength(2);
|
||||||
|
expect(useFileStore.getState().processing).toBe(true);
|
||||||
|
expect(useFileStore.getState().error).toBeNull();
|
||||||
|
expect(useFileStore.getState().activeJobId).toBe("11111111-1111-4111-8111-111111111111");
|
||||||
|
expect(useFileStore.getState().entries[0].status).toBe("processing");
|
||||||
|
|
||||||
|
act(() => {
|
||||||
|
vi.advanceTimersByTime(300_001);
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(MockEventSource.instances[1].close).toHaveBeenCalledOnce();
|
||||||
|
expect(MockEventSource.instances).toHaveLength(3);
|
||||||
|
expect(useFileStore.getState().processing).toBe(true);
|
||||||
|
|
||||||
|
act(() => {
|
||||||
|
MockEventSource.instances[2].onmessage?.({
|
||||||
|
data: JSON.stringify({
|
||||||
|
type: "single",
|
||||||
|
phase: "complete",
|
||||||
|
percent: 100,
|
||||||
|
result: {
|
||||||
|
jobId: "server-job",
|
||||||
|
downloadUrl: "/api/v1/download/server-job/upscaled.png",
|
||||||
|
originalSize: 64,
|
||||||
|
processedSize: 128,
|
||||||
|
},
|
||||||
|
}),
|
||||||
|
} as MessageEvent);
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(useFileStore.getState().processing).toBe(false);
|
||||||
|
expect(useFileStore.getState().activeJobId).toBeNull();
|
||||||
|
expect(useFileStore.getState().entries[0]).toMatchObject({
|
||||||
|
status: "completed",
|
||||||
|
processedUrl: "/api/v1/download/server-job/upscaled.png",
|
||||||
|
processedSize: 128,
|
||||||
|
});
|
||||||
|
|
||||||
|
unmount();
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user