From 4900d8a4fec9dd347d54bb4f5176a3da47440f33 Mon Sep 17 00:00:00 2001 From: SnapOtter Date: Thu, 30 Apr 2026 23:43:55 +0800 Subject: [PATCH] fix: upscale times out behind Cloudflare Tunnel due to blocking HTTP request The upscale route held the HTTP connection open for the full duration of Python sidecar processing (30-300s). Behind proxies with connection timeouts (Cloudflare Tunnel: 100s), this caused HTTP 524 errors. The route now returns 202 Accepted immediately after upload validation and processes in the background. The result (downloadUrl, sizes, etc.) is delivered via the existing SSE progress channel. The frontend detects the 202 and waits for the SSE completion event instead of reading the XHR response body. A reconnect-safe completion store ensures results survive brief SSE disconnects. Closes #106 --- apps/api/src/routes/progress.ts | 21 +++ apps/api/src/routes/tools/upscale.ts | 159 +++++++++++++---------- apps/web/src/hooks/use-tool-processor.ts | 54 +++++++- tests/integration/upscale.test.ts | 86 +++++++++--- 4 files changed, 223 insertions(+), 97 deletions(-) diff --git a/apps/api/src/routes/progress.ts b/apps/api/src/routes/progress.ts index a7cf726c..1b6cb2ce 100644 --- a/apps/api/src/routes/progress.ts +++ b/apps/api/src/routes/progress.ts @@ -32,11 +32,15 @@ export interface SingleFileProgress { stage?: string; percent: number; error?: string; + result?: Record; } /** In-memory store of job progress, keyed by jobId. */ const jobProgressStore = new Map(); +/** Terminal single-file events kept for SSE reconnect replay. */ +const singleFileCompletions = new Map(); + /** SSE listeners waiting for updates, keyed by jobId. */ const listeners = new Map void>>(); @@ -183,6 +187,16 @@ export function updateJobProgress(progress: JobProgress): void { export function updateSingleFileProgress(progress: Omit): void { const event: SingleFileProgress = { ...progress, type: "single" }; persistSingleFileProgress(progress); + + if (progress.phase === "complete" || progress.phase === "failed") { + if (singleFileCompletions.size >= 10_000) { + const oldest = singleFileCompletions.keys().next().value; + if (oldest) singleFileCompletions.delete(oldest); + } + singleFileCompletions.set(progress.jobId, event); + setTimeout(() => singleFileCompletions.delete(progress.jobId), 120_000); + } + const subs = listeners.get(progress.jobId); if (subs) { for (const cb of subs) { @@ -228,6 +242,13 @@ export async function registerProgressRoutes(app: FastifyInstance): Promise; try { - let settings: z.infer; - try { - const parsed = settingsRaw ? JSON.parse(settingsRaw) : {}; - const result = settingsSchema.safeParse(parsed); - if (!result.success) { - return reply - .status(400) - .send({ error: "Invalid settings", details: formatZodErrors(result.error.issues) }); - } - settings = result.data; - } catch { - return reply.status(400).send({ error: "Settings must be valid JSON" }); + const parsed = settingsRaw ? JSON.parse(settingsRaw) : {}; + const result = settingsSchema.safeParse(parsed); + if (!result.success) { + return reply + .status(400) + .send({ error: "Invalid settings", details: formatZodErrors(result.error.issues) }); } + settings = result.data; + } catch { + return reply.status(400).send({ error: "Settings must be valid JSON" }); + } - const scale = settings.scale; - const model = settings.model; - const faceEnhance = settings.faceEnhance; - const denoise = settings.denoise; - let format = settings.format; - const outputQuality = settings.quality; + const scale = settings.scale; + const model = settings.model; + const faceEnhance = settings.faceEnhance; + const denoise = settings.denoise; + let format = settings.format; + const outputQuality = settings.quality; + try { if (format === "auto") { const detected = await resolveOutputFormat(fileBuffer, filename); format = detected.format === "jpeg" ? "jpg" : detected.format; } - request.log.info( - { toolId: "upscale", imageSize: fileBuffer.length, scale, model, format }, - "Starting upscale", - ); // Decode HEIC/HEIF input via system decoder if (validation.format === "heif") { @@ -124,33 +120,54 @@ export function registerUpscale(app: FastifyInstance) { // Auto-orient to fix EXIF rotation before upscaling fileBuffer = await autoOrient(fileBuffer); + } catch (err) { + request.log.error({ err, toolId: "upscale" }, "Input decoding failed"); + return reply.status(422).send({ + error: "Upscaling failed", + details: err instanceof Error ? err.message : "Unknown error", + }); + } - const jobId = randomUUID(); - const workspacePath = await createWorkspace(jobId); - - // Save input + const originalSize = fileBuffer.length; + const jobId = randomUUID(); + const progressJobId = clientJobId || jobId; + let workspacePath: string; + try { + workspacePath = await createWorkspace(jobId); const inputPath = join(workspacePath, "input", filename); await writeFile(inputPath, fileBuffer); + } catch (err) { + request.log.error({ err, toolId: "upscale" }, "Workspace creation failed"); + return reply.status(422).send({ + error: "Upscaling failed", + details: err instanceof Error ? err.message : "Unknown error", + }); + } - // Determine which format the Python sidecar should produce. - // Formats that need Node.js-side conversion (HEIC/HEIF via heif-enc, - // AVIF via Sharp) are produced as PNG first, then converted below. - const needsNodeConversion = ["heic", "heif", "avif"].includes(format); - const pythonFormat = needsNodeConversion ? "png" : format; + const log = request.log; + log.info( + { toolId: "upscale", imageSize: originalSize, scale, model, format }, + "Starting upscale", + ); - // Process - const jobIdForProgress = clientJobId; - const onProgress = jobIdForProgress - ? (percent: number, stage: string) => { - updateSingleFileProgress({ - jobId: jobIdForProgress, - phase: "processing", - stage, - percent, - }); - } - : undefined; + // Reply immediately so the HTTP connection closes within proxy timeout limits. + // The result will be delivered via the SSE progress channel. + reply.status(202).send({ jobId: progressJobId, async: true }); + const needsNodeConversion = ["heic", "heif", "avif"].includes(format); + const pythonFormat = needsNodeConversion ? "png" : format; + + const onProgress = (percent: number, stage: string) => { + updateSingleFileProgress({ + jobId: progressJobId, + phase: "processing", + stage, + percent, + }); + }; + + // Fire-and-forget: processing happens after the response is sent + (async () => { const result = await upscale( fileBuffer, join(workspacePath, "output"), @@ -158,7 +175,6 @@ export function registerUpscale(app: FastifyInstance) { onProgress, ); - // Convert to final format if needed (HEIC/HEIF/AVIF) let outputBuffer = result.buffer; let finalFormat = result.format; if (needsNodeConversion) { @@ -171,7 +187,6 @@ export function registerUpscale(app: FastifyInstance) { } } - // Save output with correct extension for the chosen format const EXT_MAP: Record = { jpeg: "jpg", jpg: "jpg", @@ -188,12 +203,10 @@ export function registerUpscale(app: FastifyInstance) { const outputPath = join(workspacePath, "output", outputFilename); await writeFile(outputPath, outputBuffer); - // Generate browser-compatible preview for non-previewable formats const BROWSER_PREVIEWABLE = new Set(["png", "jpg", "jpeg", "webp", "gif", "avif", "bmp"]); let previewUrl: string | undefined; if (!BROWSER_PREVIEWABLE.has(finalFormat)) { try { - // For HEIC/HEIF, decode first since Sharp can't read HEVC const previewInput = finalFormat === "heic" || finalFormat === "heif" ? await decodeHeic(outputBuffer) @@ -203,42 +216,44 @@ export function registerUpscale(app: FastifyInstance) { await writeFile(previewPath, previewBuffer); previewUrl = `/api/v1/download/${jobId}/preview.webp`; } catch { - // Non-fatal - frontend will show fallback + // Non-fatal } } - if (clientJobId) { - updateSingleFileProgress({ - jobId: clientJobId, - phase: "complete", - percent: 100, - }); - } - if (model !== "auto" && result.method !== model) { - request.log.warn( + log.warn( { toolId: "upscale", requested: model, actual: result.method }, `Upscale model mismatch: requested ${model} but used ${result.method}`, ); } - return reply.send({ - jobId, - downloadUrl: `/api/v1/download/${jobId}/${encodeURIComponent(outputFilename)}`, - previewUrl, - originalSize: fileBuffer.length, - processedSize: outputBuffer.length, - width: result.width, - height: result.height, - method: result.method, + const downloadUrl = `/api/v1/download/${jobId}/${encodeURIComponent(outputFilename)}`; + updateSingleFileProgress({ + jobId: progressJobId, + phase: "complete", + percent: 100, + result: { + jobId, + downloadUrl, + previewUrl, + originalSize, + processedSize: outputBuffer.length, + width: result.width, + height: result.height, + method: result.method, + }, }); - } catch (err) { - request.log.error({ err, toolId: "upscale" }, "Upscaling failed"); - return reply.status(422).send({ - error: "Upscaling failed", - details: err instanceof Error ? err.message : "Unknown error", + + log.info({ toolId: "upscale", jobId, downloadUrl }, "Upscale complete"); + })().catch((err) => { + log.error({ err, toolId: "upscale" }, "Upscaling failed"); + updateSingleFileProgress({ + jobId: progressJobId, + phase: "failed", + percent: 0, + error: err instanceof Error ? err.message : "Upscale failed", }); - } + }); }); // Register in the pipeline/batch registry so this tool can be used diff --git a/apps/web/src/hooks/use-tool-processor.ts b/apps/web/src/hooks/use-tool-processor.ts index 42be10af..fe84c2aa 100644 --- a/apps/web/src/hooks/use-tool-processor.ts +++ b/apps/web/src/hooks/use-tool-processor.ts @@ -94,6 +94,7 @@ export function useToolProcessor(toolId: string) { // Generate client job ID for SSE correlation const clientJobId = generateId(); + let asyncMode = false; // For AI tools, open SSE before uploading if (isAiTool) { @@ -104,8 +105,42 @@ export function useToolProcessor(toolId: string) { es.onmessage = (event) => { try { const data = JSON.parse(event.data); - if (data.type === "single" && typeof data.percent === "number") { - // Scale server progress (0-100) into 15-100 range + if (data.type !== "single") return; + + if (data.phase === "complete" && data.result) { + if (elapsedRef.current) clearInterval(elapsedRef.current); + if (processingTimerRef.current) clearInterval(processingTimerRef.current); + es.close(); + eventSourceRef.current = null; + + const result = data.result as ProcessResult; + setWarning(result.warning ?? null); + useFileStore.getState().updateEntry(capturedIndex, { + processedUrl: result.downloadUrl, + processedPreviewUrl: result.previewUrl ?? null, + processedFilename: null, + status: "completed", + originalSize: result.originalSize, + processedSize: result.processedSize, + ...(result.savedFileId ? { serverFileId: result.savedFileId } : {}), + }); + setProcessing(false); + setProgress(IDLE_PROGRESS); + return; + } + + if (data.phase === "failed" && asyncMode) { + if (elapsedRef.current) clearInterval(elapsedRef.current); + if (processingTimerRef.current) clearInterval(processingTimerRef.current); + es.close(); + eventSourceRef.current = null; + setError(data.error || "Processing failed"); + setProcessing(false); + setProgress(IDLE_PROGRESS); + return; + } + + if (typeof data.percent === "number") { const scaled = 15 + (data.percent / 100) * 85; setProgress((prev) => ({ ...prev, @@ -120,11 +155,13 @@ export function useToolProcessor(toolId: string) { }; es.onerror = () => { - es.close(); - eventSourceRef.current = null; + if (!asyncMode) { + es.close(); + eventSourceRef.current = null; + } }; } catch { - // EventSource creation failed — proceed without SSE + // EventSource creation failed -- proceed without SSE } } @@ -209,6 +246,11 @@ export function useToolProcessor(toolId: string) { }; xhr.onload = () => { + if (xhr.status === 202) { + asyncMode = true; + return; + } + if (elapsedRef.current) clearInterval(elapsedRef.current); if (processingTimerRef.current) clearInterval(processingTimerRef.current); if (eventSourceRef.current) { @@ -220,8 +262,6 @@ export function useToolProcessor(toolId: string) { try { const result: ProcessResult = JSON.parse(xhr.responseText); setWarning(result.warning ?? null); - // Write result to the entry that was being processed (captured at - // request time), not whatever entry happens to be selected now. useFileStore.getState().updateEntry(capturedIndex, { processedUrl: result.downloadUrl, processedPreviewUrl: result.previewUrl ?? null, diff --git a/tests/integration/upscale.test.ts b/tests/integration/upscale.test.ts index 8b6ec131..3c085148 100644 --- a/tests/integration/upscale.test.ts +++ b/tests/integration/upscale.test.ts @@ -1,9 +1,10 @@ /** * Integration tests for the upscale tool (/api/v1/tools/upscale). * - * This tool requires the Python sidecar (Real-ESRGAN). Tests accept both - * 200 (sidecar running) and 501 (not installed) for the processing path - * while fully testing validation paths. + * This tool uses async processing: valid requests return 202 with a jobId, + * and the result is delivered via SSE. Tests accept both 202 (processing + * accepted) and 501 (not installed) for the processing path while fully + * testing validation paths. */ import { readFileSync } from "node:fs"; @@ -50,7 +51,7 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); }, 60_000); it("accepts default settings (2x scale)", async () => { @@ -69,15 +70,12 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); - if (res.statusCode === 200) { + if (res.statusCode === 202) { const result = JSON.parse(res.body); - expect(result.downloadUrl).toBeDefined(); - expect(result.processedSize).toBeGreaterThan(0); - expect(result.width).toBeDefined(); - expect(result.height).toBeDefined(); - expect(result.method).toBeDefined(); + expect(result.jobId).toBeDefined(); + expect(result.async).toBe(true); } if (res.statusCode === 501) { @@ -105,7 +103,7 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); }, 60_000); it("accepts model and faceEnhance options", async () => { @@ -131,7 +129,7 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); }, 60_000); it("accepts denoise and format options", async () => { @@ -158,7 +156,7 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); }, 60_000); it("accepts scale as a string (coerced to number)", async () => { @@ -180,7 +178,7 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); }, 60_000); it("processes JPEG input", async () => { @@ -199,7 +197,7 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); }, 60_000); it("handles HEIC input", async () => { @@ -218,7 +216,7 @@ describe("Upscale", () => { body, }); - expect([200, 501]).toContain(res.statusCode); + expect([202, 501]).toContain(res.statusCode); }, 60_000); it("handles 1x1 pixel input", async () => { @@ -237,7 +235,7 @@ describe("Upscale", () => { body, }); - expect([200, 422, 501]).toContain(res.statusCode); + expect([202, 422, 501]).toContain(res.statusCode); }, 60_000); // ── Validation (always testable) ───────────────────────────────── @@ -302,4 +300,56 @@ describe("Upscale", () => { expect(res.statusCode).toBe(401); }); + + // ── Async processing (regression for #106) ───────────────────────── + + it("returns 202 with jobId for async processing", async () => { + const { body, contentType } = createMultipartPayload([ + { name: "file", filename: "test.png", contentType: "image/png", content: PNG }, + { name: "settings", content: JSON.stringify({ scale: 2 }) }, + { name: "clientJobId", content: "test-job-async-regression" }, + ]); + + const res = await app.inject({ + method: "POST", + url: "/api/v1/tools/upscale", + headers: { + authorization: `Bearer ${adminToken}`, + "content-type": contentType, + }, + body, + }); + + if (res.statusCode === 501) return; + + expect(res.statusCode).toBe(202); + const result = JSON.parse(res.body); + expect(result.async).toBe(true); + expect(result.jobId).toBe("test-job-async-regression"); + }, 60_000); + + it("returns 202 without blocking for processing", async () => { + const { body, contentType } = createMultipartPayload([ + { name: "file", filename: "test.png", contentType: "image/png", content: PNG }, + { name: "settings", content: JSON.stringify({}) }, + { name: "clientJobId", content: "test-job-timing" }, + ]); + + const start = Date.now(); + const res = await app.inject({ + method: "POST", + url: "/api/v1/tools/upscale", + headers: { + authorization: `Bearer ${adminToken}`, + "content-type": contentType, + }, + body, + }); + + if (res.statusCode === 501) return; + + const elapsed = Date.now() - start; + expect(res.statusCode).toBe(202); + expect(elapsed).toBeLessThan(30_000); + }, 60_000); });