From ed25603be24a310c6b9080b737b57ccfcc5e941a Mon Sep 17 00:00:00 2001 From: WuYongJie <549727955@qq.com> Date: Fri, 13 Nov 2026 18:39:43 +0800 Subject: [PATCH] fix: prepare CDN originals without download relay fallback --- app/api/canvas/download-image/route.ts | 23 ++-- app/api/image-proxy/route.ts | 4 +- lib/client/canvas-image-download.ts | 25 +++- lib/jobs/image-archive-job.ts | 74 ++++++++---- lib/storage/canvas-cdn-download.ts | 2 + lib/storage/image-archive-db.ts | 26 ++++ lib/storage/image-archive-worker.ts | 120 +++++++++++-------- lib/storage/image-archive.ts | 7 +- lib/storage/original-image-download.ts | 31 +++-- playwright.helpers.config.ts | 1 + tests/routes/image-archive-flow.test.ts | 38 ++---- tests/routes/image-archive-lanes.test.ts | 67 +++++++++++ tests/routes/original-image-download.test.ts | 66 +++++----- 13 files changed, 314 insertions(+), 170 deletions(-) create mode 100644 tests/routes/image-archive-lanes.test.ts diff --git a/app/api/canvas/download-image/route.ts b/app/api/canvas/download-image/route.ts index a17b6d858..5d69a7566 100644 --- a/app/api/canvas/download-image/route.ts +++ b/app/api/canvas/download-image/route.ts @@ -26,9 +26,10 @@ function failed(failure: DownloadFailure, requestId: string) { }); } -function preparing(requestId: string) { +function preparing(requestId: string, converting = false) { return NextResponse.json({ delivery: "preparing", retryAfter: 2, requestId, - message: "正在准备原图，完成后将自动开始下载" }, { + phase: converting ? "converting" : "preparing", + message: converting ? "正在转换为兼容 PNG 并保存，完成后自动下载" : "正在准备原图，完成后将自动开始下载" }, { status: 202, headers: { "Retry-After": "2", "Cache-Control": "no-store" }, }); } @@ -119,23 +120,23 @@ export async function POST(request: NextRequest) { const name = sanitizeCanvasDownloadFileName(String(body.name || ""), "canvas-image"); if (body.preferDirect === true) { - const { task, direct, canPrepareDirect } = await resolveCanvasCdnDownload(urls, `${name}.png`).catch((error) => { - logger.warn("cdn_lookup_failed", error, { requestId }); - return { task: null, direct: null, canPrepareDirect: false }; - }); + const { task, direct, canPrepareDirect } = await resolveCanvasCdnDownload(urls, `${name}.png`); if (direct) return NextResponse.json(direct, { headers: { "Cache-Control": "no-store" } }); - if (task && canPrepareDirect && body.preparation === "async") { + if (task && canPrepareDirect) { const asset = (task.metadata as { downloadAsset?: { status?: string; failure?: DownloadFailure; retryAt?: string } } | null)?.downloadAsset; if (asset?.status === "abandoned" || (asset?.status === "failed" && Date.parse(asset.retryAt || "") > Date.now())) { return failed(asset.failure || getDownloadFailure(null), task.id); } scheduleImageDownloadPreparation(task.id); - return preparing(task.id); + return preparing(task.id, true); } - if (task?.targetUrl) { - if (!urls.includes(task.targetUrl)) urls.unshift(task.targetUrl); - scheduleImageDownloadPreparation(task.id); + if (task && !task.targetUrl) { + const { originalImagePreparing, prepareOriginalImageArchive } = await import("@/lib/storage/original-image-download"); + const pending = await prepareOriginalImageArchive(task.sourceUrl); + return NextResponse.json(originalImagePreparing(pending), { status: 202, + headers: { "Retry-After": "3", "Cache-Control": "no-store" } }); } + throw new Error("下载文件暂时无法从存储桶直接下载"); } const cookie = request.headers.get("cookie") || ""; if (body.preparation === "async") { diff --git a/app/api/image-proxy/route.ts b/app/api/image-proxy/route.ts index 29d2e8cf9..2a8097d9c 100644 --- a/app/api/image-proxy/route.ts +++ b/app/api/image-proxy/route.ts @@ -367,7 +367,9 @@ export async function GET(request: NextRequest) { buildImageProxyUrl(target.toString(), { delivery: "original" }); const { resolveOriginalCdnDownload } = await import("@/lib/storage/original-image-download"); const direct = await resolveOriginalCdnDownload(target.toString(), searchParams.get("name") || "", request.signal); - return NextResponse.json(direct || { delivery: "proxy" }, { headers: { "Cache-Control": "no-store" } }); + if (!direct) throw new Error("下载文件暂时无法从存储桶直接下载"); + return NextResponse.json(direct, { status: direct.delivery === "preparing" ? 202 : 200, + headers: { "Cache-Control": "no-store", ...(direct.delivery === "preparing" ? { "Retry-After": "3" } : {}) } }); } catch (error) { return downloadFailure(error); } diff --git a/lib/client/canvas-image-download.ts b/lib/client/canvas-image-download.ts index 51a97e4c9..8d2c60ed2 100644 --- a/lib/client/canvas-image-download.ts +++ b/lib/client/canvas-image-download.ts @@ -201,7 +201,8 @@ async function downloadOriginalImageFile(params: DownloadCanvasImageParams): Pro const { url, name, signal, onProgress } = params; const source = String(url || "").trim(); if (!source) throw new Error("图片地址无效"); - const proxy = buildImageProxyUrl(source, { delivery: "original" }); + const sourceUrl = new URL(source, window.location.origin); + const proxy = buildImageProxyUrl(/^https?:$/.test(sourceUrl.protocol) ? sourceUrl.toString() : source, { delivery: "original" }); const filenameBase = sanitizeDownloadName(name || inferDownloadNameFromUrl(source), "image"); const target = new URL(proxy, window.location.origin); onProgress?.({ phase: "preparing", message: "正在准备原图..." }); @@ -223,8 +224,20 @@ async function downloadOriginalImageFile(params: DownloadCanvasImageParams): Pro if (target.origin === window.location.origin && target.pathname === "/api/image-proxy") { target.searchParams.set("download", "resolve"); target.searchParams.set("name", filenameBase); - const metadata = await checkedFetch(target.pathname + target.search); - const delivery = await metadata.json(); + const deadline = Date.now() + 180_000; + let delivery: any; + for (let poll = 0; ; poll++) { + signal?.throwIfAborted(); + const metadata = await checkedFetch(target.pathname + target.search); + delivery = await metadata.json(); + if (metadata.status !== 202) break; + if (delivery.delivery !== "preparing") throw new Error("下载准备状态无效"); + if (Date.now() >= deadline || poll >= 60) { + throw new CanvasImageDownloadError("原图仍在准备中", { code: "preparation_timeout", requestId: delivery.requestId }); + } + onProgress?.({ phase: "preparing", message: delivery.message || "正在准备原图，完成后自动下载" }); + await waitForRetry(Math.max(1, Math.min(5, Number(delivery.retryAfter) || 3)) * 1000, signal); + } if (delivery.delivery === "cdn") { // The resolver names the attachment using the stored MIME type. const extension = delivery.filename.split(".").pop(); @@ -235,7 +248,7 @@ async function downloadOriginalImageFile(params: DownloadCanvasImageParams): Pro onProgress?.({ phase: "downloading", message: "已开始下载，可在浏览器下载列表查看进度" }); return result; } - if (delivery.delivery !== "proxy") throw new Error("下载链接无效"); + throw new CanvasImageDownloadError("原图尚未准备完成", { code: "preparation_timeout", requestId: delivery.requestId }); } const response = await checkedFetch(proxy); const contentType = response.headers.get("content-type") || ""; @@ -377,8 +390,8 @@ async function downloadPngFile({ throw new CanvasImageDownloadError("原图仍在准备中", { code: "preparation_timeout", status: 202, requestId: preparing.requestId }); } onProgress?.({ - phase: preparationPhase, - message: isConverting ? conversionMessage : preparing.message || "正在准备原图，完成后将自动下载", + phase: preparing.phase || preparationPhase, + message: preparing.message || (isConverting ? conversionMessage : "正在准备原图，完成后将自动下载"), }); const delaySeconds = Math.max(preparationPolls > 5 ? 5 : 1, Math.min(5, Number(preparing.retryAfter) || 3)); await waitForRetry(delaySeconds * 1000, signal); diff --git a/lib/jobs/image-archive-job.ts b/lib/jobs/image-archive-job.ts index c826680da..d8db1ae8e 100644 --- a/lib/jobs/image-archive-job.ts +++ b/lib/jobs/image-archive-job.ts @@ -15,11 +15,56 @@ import { markImageArchiveTaskReplaceFailed, markImageArchiveTaskReplacing, markImageArchiveTaskStored, + claimNextImageArchiveTask, + type ImageArchiveTaskRecord, } from "@/lib/storage/image-archive-db"; import { replaceImageUrlInDatabase } from "@/lib/storage/image-url-replacement"; const DEFAULT_BATCH_SIZE = 10; +async function storeClaimedImage(task: ImageArchiveTaskRecord) { + try { + const metadata = task.metadata as { inlineSource?: string } | null; + const archived = await archiveTemporaryImage(metadata?.inlineSource || task.sourceUrl, task.id); + await markImageArchiveTaskStored({ id: task.id, targetUrl: archived.url, downloadAsset: archived.downloadAsset }); + if (!archived.downloadAsset) scheduleImageDownloadPreparation(task.id); + console.log("[ImageArchiveJob] stored", { taskId: task.id, bytes: archived.bytes, targetUrl: archived.url }); + return true; + } catch (error) { + await markImageArchiveTaskFailed({ id: task.id, error }); + console.error("[ImageArchiveJob] store failed", { taskId: task.id, error: String(error) }); + return false; + } +} + +export async function storeNextImageArchive(preferredTaskId?: string) { + return trackRuntimeWork(async () => { + const task = await claimNextImageArchiveTask(preferredTaskId); + if (!task) return false; + await storeClaimedImage(task); + return true; + }, { durable: true }); +} + +export async function replacePendingImageArchives() { + return trackRuntimeWork(async () => { + const tasks = await getImageArchiveTasksToReplace({ limit: 10 }); + let claimed = 0; + for (const task of tasks) { + if (!(await markImageArchiveTaskReplacing(task.id))) continue; + claimed++; + try { + if (!task.targetUrl) throw new Error("Missing targetUrl"); + const result = await replaceImageUrlInDatabase(task.sourceUrl, task.targetUrl); + await markImageArchiveTaskCompleted({ id: task.id, replacedCount: result.changed }); + } catch (error) { + await markImageArchiveTaskReplaceFailed({ id: task.id, error }); + } + } + return claimed; + }, { durable: true }); +} + function getBatchSize() { const parsed = Number(process.env.IMAGE_ARCHIVE_BATCH_SIZE || DEFAULT_BATCH_SIZE); if (!Number.isFinite(parsed)) return DEFAULT_BATCH_SIZE; @@ -38,32 +83,9 @@ export async function runImageArchiveJob({ preferredTaskId }: { preferredTaskId? let storeFailed = 0; const storeOne = async (task: (typeof storeTasks)[number]) => { - try { - if (!(await markImageArchiveTaskArchiving(task.id))) return; - const metadata = task.metadata as { inlineSource?: string } | null; - const archived = await archiveTemporaryImage(metadata?.inlineSource || task.sourceUrl, task.id); - await markImageArchiveTaskStored({ - id: task.id, - targetUrl: archived.url, - downloadAsset: archived.downloadAsset, - }); - if (!archived.downloadAsset) scheduleImageDownloadPreparation(task.id); - storeSucceeded += 1; - console.log("[ImageArchiveJob] stored", { - taskId: task.id, - bytes: archived.bytes, - sourceUrl: task.sourceUrl, - targetUrl: archived.url, - }); - } catch (error) { - storeFailed += 1; - await markImageArchiveTaskFailed({ id: task.id, error }); - console.error("[ImageArchiveJob] store failed", { - taskId: task.id, - sourceUrl: task.sourceUrl, - error: error instanceof Error ? error.message : String(error), - }); - } + if (!(await markImageArchiveTaskArchiving(task.id))) return; + if (await storeClaimedImage(task)) storeSucceeded++; + else storeFailed++; }; // Bucket work has a small worker pool independent of unrestricted model generation. diff --git a/lib/storage/canvas-cdn-download.ts b/lib/storage/canvas-cdn-download.ts index b3d698917..1029253ff 100644 --- a/lib/storage/canvas-cdn-download.ts +++ b/lib/storage/canvas-cdn-download.ts @@ -2,6 +2,7 @@ import "server-only"; import { getDefaultStorageConfig } from "@/lib/db/queries"; import { ensureStoredImageDownloadTask, findImageArchiveDownloadTask } from "./image-archive-db"; import { extractImageProxySourceUrl, inferOriginalImageUrl } from "@/lib/image-url"; +import { prepareOriginalImageArchive } from "./original-image-download"; export async function resolveCanvasCdnDownload(urls: string[], filename: string) { const sources = Array.from(new Set(urls.flatMap(url => { @@ -27,6 +28,7 @@ export async function resolveCanvasCdnDownload(urls: string[], filename: string) } } } + if (!task) task = await prepareOriginalImageArchive(sources[0]); const asset = (task?.metadata as { downloadAsset?: { status?: string; url?: string; contentType?: string; bytes?: number } } | null)?.downloadAsset; const ready = asset?.status === "ready" && asset.url && asset.contentType === "image/png"; const source = ready ? asset.url : task?.targetUrl; diff --git a/lib/storage/image-archive-db.ts b/lib/storage/image-archive-db.ts index 7484f6b52..d074b5707 100644 --- a/lib/storage/image-archive-db.ts +++ b/lib/storage/image-archive-db.ts @@ -137,6 +137,32 @@ export async function markImageArchiveTaskArchiving(id: string) { return claimed.length > 0; } +export async function claimNextImageArchiveTask(preferredTaskId?: string) { + return db.transaction(async (tx) => { + const priority = preferredTaskId + ? [asc(sql`case when ${imageArchiveTask.id} = ${preferredTaskId} then 0 else 1 end`)] : []; + const [task] = await tx.select().from(imageArchiveTask).where(and( + inArray(imageArchiveTask.status, ["pending", "failed"]), + lte(imageArchiveTask.archiveAfterAt, new Date()), + lt(imageArchiveTask.attemptCount, MAX_IMAGE_ARCHIVE_ATTEMPTS), + )).orderBy(...priority, asc(imageArchiveTask.createdAt), asc(imageArchiveTask.id)) + .limit(1).for("update", { skipLocked: true }); + if (!task) return null; + await tx.update(imageArchiveTask).set({ status: "archiving", + attemptCount: sql`${imageArchiveTask.attemptCount} + 1`, lastError: null, updatedAt: new Date(), + }).where(eq(imageArchiveTask.id, task.id)); + return task; + }); +} + +export async function hasPendingImageArchives() { + const [task] = await db.select({ id: imageArchiveTask.id }).from(imageArchiveTask).where(or( + inArray(imageArchiveTask.status, ["pending", "failed", "archiving", "stored", "replacing"]), + and(eq(imageArchiveTask.status, "completed"), sql`${imageArchiveTask.metadata}->>'replacementPasses' = '1'`), + )).limit(1); + return Boolean(task); +} + export async function markImageArchiveTaskStored(params: { id: string; targetUrl: string; diff --git a/lib/storage/image-archive-worker.ts b/lib/storage/image-archive-worker.ts index 51230b62a..e001464bb 100644 --- a/lib/storage/image-archive-worker.ts +++ b/lib/storage/image-archive-worker.ts @@ -1,70 +1,88 @@ import "server-only"; import { after } from "next/server"; -const workerKey = "__generated_image_archive_worker__"; +const workerKey = "__generated_image_archive_worker_v2__"; type Worker = { - running: boolean; + active: number; scheduled: boolean; - requested: boolean; - preferredTaskId?: string; + priority: Set; + replacing: boolean; + replaceRequested: boolean; retryTimer?: ReturnType; }; function getWorker(): Worker { const root = globalThis as typeof globalThis & { [workerKey]?: Worker }; - return (root[workerKey] ??= { running: false, scheduled: false, requested: false }); + return (root[workerKey] ??= { active: 0, scheduled: false, priority: new Set(), replacing: false, replaceRequested: false }); +} + +async function retryIfNeeded() { + const worker = getWorker(); + if (worker.active || worker.replacing || worker.retryTimer) return; + const { hasPendingImageArchives, hasPendingImageDownloads } = await import("./image-archive-db"); + if (!(await hasPendingImageArchives()) && !(await hasPendingImageDownloads())) return; + if (worker.retryTimer) return; + worker.retryTimer = setTimeout(() => { + worker.retryTimer = undefined; + scheduleImageArchiveWorker(); + }, 60_000); + worker.retryTimer.unref?.(); +} + +function scheduleReplacement() { + const worker = getWorker(); + worker.replaceRequested = true; + if (worker.replacing) return; + worker.replacing = true; + void (async () => { + const { replacePendingImageArchives } = await import("@/lib/jobs/image-archive-job"); + do { + worker.replaceRequested = false; + while (await replacePendingImageArchives()) { /* Drain references independently of uploads. */ } + } while (worker.replaceRequested); + })().catch(error => console.error("[ImageArchiveWorker] replacement failed", error)).finally(() => { + worker.replacing = false; + void retryIfNeeded().catch(error => console.error("[ImageArchiveWorker] retry failed", error)); + }); } export function scheduleImageArchiveWorker(deliveryConfirmed = false, preferredTaskId?: string) { const worker = getWorker(); - worker.preferredTaskId ??= preferredTaskId; - worker.requested = true; - if (worker.running || (worker.scheduled && !deliveryConfirmed)) return; + if (preferredTaskId) worker.priority.add(preferredTaskId); + if (worker.scheduled && !deliveryConfirmed) return; worker.scheduled = true; - if (worker.retryTimer) { - clearTimeout(worker.retryTimer); - worker.retryTimer = undefined; - } const run = async () => { - if (worker.running) return; + if (!worker.scheduled) return; worker.scheduled = false; - worker.running = true; - let retryNeeded = false; - try { - const { runImageArchiveJob } = await import("@/lib/jobs/image-archive-job"); - do { - worker.requested = false; - const nextTaskId = worker.preferredTaskId; - worker.preferredTaskId = undefined; - const result = await runImageArchiveJob({ preferredTaskId: nextTaskId }); - retryNeeded = result.stats.pendingStore > 0 || result.stats.pendingReplace > 0 || result.pendingDownloads === true; - // Stop when only backoff tasks remain; do not spin while storage is down. - if (!worker.requested && result.store.total === 0 && result.replace.total === 0 && result.downloadTasks === 0) break; - } while (true); - } catch (error) { - retryNeeded = true; - console.error("[ImageArchiveWorker] background archive failed", error); - } finally { - worker.running = false; - if (worker.requested) scheduleImageArchiveWorker(); - else if (retryNeeded) { - worker.retryTimer = setTimeout(scheduleImageArchiveWorker, 60_000); - worker.retryTimer.unref?.(); - } + if (worker.retryTimer) { clearTimeout(worker.retryTimer); worker.retryTimer = undefined; } + const { storeNextImageArchive } = await import("@/lib/jobs/image-archive-job"); + const { recoverStaleImageArchiveTasks } = await import("./image-archive-db"); + const { scheduleImageDownloadPreparation } = await import("./image-download-preparation"); + await recoverStaleImageArchiveTasks(); + scheduleImageDownloadPreparation(); + scheduleReplacement(); + // Free slots take the next file immediately; slow files and reference updates + // cannot hold an entire batch or delay newly generated images. + while (worker.active < 4) { + worker.active++; + void (async () => { + for (;;) { + const id = worker.priority.values().next().value; + if (id) worker.priority.delete(id); + if (!(await storeNextImageArchive(id))) break; + scheduleReplacement(); + } + })().catch(error => console.error("[ImageArchiveWorker] storage failed", error)).finally(() => { + worker.active--; + if (worker.priority.size) scheduleImageArchiveWorker(); + else void retryIfNeeded().catch(error => console.error("[ImageArchiveWorker] retry failed", error)); + }); } }; - // Detached generation tasks may inherit a request that has already closed. Keep - // a Node-server fallback so archiving never depends on the user leaving a tab open. - const deliveryFallback = setTimeout(() => { - if (worker.scheduled && !worker.running) void run(); - }, 1_000); - deliveryFallback.unref?.(); - try { - // Request handlers deliver first. Durable task rows also allow cron/restart recovery. - after(run); - } catch { - // Durable canvas tasks can run outside Next's request context on the Node server. - setImmediate(() => { - void run(); - }); - } + const safeRun = () => run().catch(error => { + console.error("[ImageArchiveWorker] scheduling failed", error); + void retryIfNeeded().catch(error => console.error("[ImageArchiveWorker] retry failed", error)); + }); + const fallback = setTimeout(() => { if (worker.scheduled) void safeRun(); }, 1_000); + fallback.unref?.(); + try { after(safeRun); } catch { setImmediate(safeRun); } } diff --git a/lib/storage/image-archive.ts b/lib/storage/image-archive.ts index a634e6e1f..26b31e9f0 100644 --- a/lib/storage/image-archive.ts +++ b/lib/storage/image-archive.ts @@ -58,7 +58,8 @@ export async function registerGeneratedImageTemporaryUrls(params: { displayHoldMs: getImageArchiveDisplayHoldMs(), }, }); - scheduleImageArchiveWorker(); + for (const task of tasks) scheduleImageArchiveWorker(false, task.id); + if (!tasks.length) scheduleImageArchiveWorker(); return tasks; } catch (error) { logger.warn("enqueue temporary image archive failed", { @@ -199,10 +200,10 @@ export async function archiveTemporaryImage(sourceUrl: string, archiveId?: strin }); if (direct) return direct; } - const downloaded = await downloadTemporaryImage({ sourceUrl }); + const downloaded = await downloadTemporaryImage({ sourceUrl, retries: 0 }); const objectSuffix = `${Date.now()}-${randomUUID()}`; const filename = `ai-generated-archive-${objectSuffix}.${downloaded.extension}`; - const uploaded = await uploadFile(filename, toArrayBuffer(downloaded.buffer)); + const uploaded = await uploadFile(filename, toArrayBuffer(downloaded.buffer), { preferQiniuForm: true }); await cacheImageBytes(`source::${uploaded.url}`, downloaded); return { url: uploaded.url, diff --git a/lib/storage/original-image-download.ts b/lib/storage/original-image-download.ts index e879bb60e..0fcf40f66 100644 --- a/lib/storage/original-image-download.ts +++ b/lib/storage/original-image-download.ts @@ -1,11 +1,31 @@ import "server-only"; import { getDefaultStorageConfig } from "@/lib/db/queries"; -import { findImageArchiveDownloadTask } from "./image-archive-db"; +import { enqueueImageArchiveTask, findImageArchiveDownloadTask } from "./image-archive-db"; import { inferOriginalImageUrl } from "@/lib/image-url"; import { getImageDownloadExtension } from "@/lib/image-delivery"; import { sanitizeCanvasDownloadFileName } from "@/lib/canvas/photoshop-safe-download"; import { HeadObjectCommand } from "@aws-sdk/client-s3"; +export async function prepareOriginalImageArchive(source: string) { + const original = inferOriginalImageUrl(source) || source; + const task = await findImageArchiveDownloadTask([...new Set([source, original])]) + || await enqueueImageArchiveTask({ sourceUrl: original, metadata: { requestedDownload: true } }); + if (!task) throw new Error("下载文件尚未保存到存储桶"); + if (!task.targetUrl) { + if (task.status === "abandoned" || (task.status === "failed" && task.archiveAfterAt > new Date())) { + throw new Error(task.lastError || "下载文件保存到存储桶失败"); + } + const { scheduleImageArchiveWorker } = await import("./image-archive-worker"); + scheduleImageArchiveWorker(false, task.id); + } + return task; +} + +export function originalImagePreparing(task: { id: string; status: string }) { + return { delivery: "preparing" as const, phase: "preparing" as const, retryAfter: 3, requestId: task.id, + message: task.status === "archiving" ? "正在保存原图，完成后自动下载" : "正在排队准备原图，完成后自动下载" }; +} + /** Resolve the archived original without waiting for a separate PNG conversion. */ export async function resolveOriginalCdnDownload(source: string, name: string, signal: AbortSignal) { const config = await getDefaultStorageConfig(); @@ -14,12 +34,9 @@ export async function resolveOriginalCdnDownload(source: string, name: string, s const origin = process.env.CANVAS_QINIU_ORIGIN_URL; let target = new URL(source); if (target.origin !== base.origin && (!origin || target.origin !== new URL(origin).origin)) { - const task = await findImageArchiveDownloadTask([...new Set([source, inferOriginalImageUrl(source)])]); - if (task?.status === "pending" && !task.targetUrl) { - const { scheduleImageArchiveWorker } = await import("./image-archive-worker"); - scheduleImageArchiveWorker(false, task.id); - } - target = new URL(task?.targetUrl || source); + const task = await prepareOriginalImageArchive(source); + if (!task.targetUrl) return originalImagePreparing(task); + target = new URL(task.targetUrl); } if (origin && target.origin === new URL(origin).origin && !target.searchParams.has("token")) { target.host = base.host; diff --git a/playwright.helpers.config.ts b/playwright.helpers.config.ts index a90588d07..c673df6ee 100644 --- a/playwright.helpers.config.ts +++ b/playwright.helpers.config.ts @@ -12,6 +12,7 @@ export default defineConfig({ /routes\/canvas-download-rate-limit\.test\.ts/, /routes\/canvas-zip-download\.test\.ts/, /routes\/image-archive-flow\.test\.ts/, + /routes\/image-archive-lanes\.test\.ts/, /routes\/runtime-active-work\.test\.ts/, /routes\/canvas-capability-task-recovery\.test\.ts/, /routes\/canvas-toolbar-capability\.test\.ts/, diff --git a/tests/routes/image-archive-flow.test.ts b/tests/routes/image-archive-flow.test.ts index 55ece594f..b543146ae 100644 --- a/tests/routes/image-archive-flow.test.ts +++ b/tests/routes/image-archive-flow.test.ts @@ -244,21 +244,12 @@ test("background storage starts after delivery and coalesces repeated wakeups", const callbacks: Array<() => Promise> = []; const preferences: Array = []; const module = serverLoader({ - "next/server": { - after: (fn: () => Promise) => { - callbacks.push(fn); - }, - }, + "next/server": { after: (fn: () => Promise) => callbacks.push(fn) }, + "./image-archive-db": { recoverStaleImageArchiveTasks: async () => {}, hasPendingImageArchives: async () => false, hasPendingImageDownloads: async () => false }, + "./image-download-preparation": { scheduleImageDownloadPreparation() {} }, "@/lib/jobs/image-archive-job": { - runImageArchiveJob: async ({ preferredTaskId }: { preferredTaskId?: string }) => { - preferences.push(preferredTaskId); - return { - store: { total: preferences.length === 1 ? 1 : 0 }, - replace: { total: 0 }, - stats: { pendingStore: 0, pendingReplace: 0 }, - downloadTasks: 0, - }; - }, + storeNextImageArchive: async (id?: string) => { preferences.push(id); return false; }, + replacePendingImageArchives: async () => 0, }, })("lib/storage/image-archive-worker.ts"); module.scheduleImageArchiveWorker(); @@ -266,7 +257,8 @@ test("background storage starts after delivery and coalesces repeated wakeups", expect(callbacks).toHaveLength(1); expect(preferences).toEqual([]); await callbacks[0](); - expect(preferences).toEqual(["requested-image", undefined]); + expect(preferences[0]).toBe("requested-image"); + expect(preferences).toHaveLength(4); }); test("bucket failure retains the provider URL and retry task", async () => { @@ -662,21 +654,15 @@ test("replacement retries reconcile late writes once, without reclaiming complet test("detached generation still archives when its original request has already closed", async () => { let runs = 0; const module = serverLoader({ - // A closed request may accept a callback that is never invoked. "next/server": { after: () => {} }, + "./image-archive-db": { recoverStaleImageArchiveTasks: async () => {}, hasPendingImageArchives: async () => false, hasPendingImageDownloads: async () => false }, + "./image-download-preparation": { scheduleImageDownloadPreparation() {} }, "@/lib/jobs/image-archive-job": { - runImageArchiveJob: async () => { - runs += 1; - return { - store: { total: 0 }, - replace: { total: 0 }, - stats: { pendingStore: 0, pendingReplace: 0 }, - downloadTasks: 0, - }; - }, + storeNextImageArchive: async () => { runs++; return false; }, + replacePendingImageArchives: async () => 0, }, })("lib/storage/image-archive-worker.ts"); module.scheduleImageArchiveWorker(); expect(runs).toBe(0); - await expect.poll(() => runs, { timeout: 5_000 }).toBe(1); + await expect.poll(() => runs, { timeout: 5_000 }).toBe(4); }); diff --git a/tests/routes/image-archive-lanes.test.ts b/tests/routes/image-archive-lanes.test.ts new file mode 100644 index 000000000..8f6997149 --- /dev/null +++ b/tests/routes/image-archive-lanes.test.ts @@ -0,0 +1,67 @@ +import { test, expect } from "@playwright/test"; +import { serverLoader } from "../helpers/canvas-download-server-loader"; + +test("a slow original and slow reference updates do not block later files or new downloads", async () => { + let finishSlow!: () => void; + let finishReplacement!: () => void; + const slow = new Promise(resolve => { finishSlow = resolve; }); + const replacement = new Promise(resolve => { finishReplacement = resolve; }); + const pending = Array.from({ length: 25 }, (_, i) => `image-${i}`); + const completed: string[] = []; + const callbacks: Array<() => Promise> = []; + let active = 0; + let peak = 0; + let replacing = false; + const module = serverLoader({ + "next/server": { after: (callback: () => Promise) => callbacks.push(callback) }, + "./image-archive-db": { recoverStaleImageArchiveTasks: async () => {}, hasPendingImageArchives: async () => false, hasPendingImageDownloads: async () => false }, + "./image-download-preparation": { scheduleImageDownloadPreparation() {} }, + "@/lib/jobs/image-archive-job": { + storeNextImageArchive: async (preferred?: string) => { + const index = preferred ? pending.indexOf(preferred) : -1; + const id = pending.splice(index < 0 ? 0 : index, 1)[0]; + if (!id) return false; + active++; peak = Math.max(peak, active); + if (id === "image-0") await slow; + else await new Promise(resolve => setTimeout(resolve, 1)); + completed.push(id); active--; + return true; + }, + replacePendingImageArchives: async () => { replacing = true; await replacement; return 0; }, + }, + })("lib/storage/image-archive-worker.ts"); + module.scheduleImageArchiveWorker(); + await callbacks.shift()!(); + await expect.poll(() => completed.length).toBe(24); + expect(completed).not.toContain("image-0"); + expect(replacing).toBe(true); + pending.push("new-download"); + module.scheduleImageArchiveWorker(false, "new-download"); + await callbacks.shift()!(); + await expect.poll(() => completed.includes("new-download")).toBe(true); + expect(peak).toBeLessThanOrEqual(4); + finishSlow(); finishReplacement(); + await expect.poll(() => completed.length).toBe(26); + expect(new Set(completed).size).toBe(26); +}); + +test("conversion requests wait for the original archive and never invoke the relay preparation queue", async () => { + const pending = { id: "queued", sourceUrl: "https://model.example/image", status: "pending" }; + const route = serverLoader({ + "next/server": { NextResponse: Response }, + "@/app/(auth)/auth": { auth: async () => ({ user: { id: "test" } }) }, + "@/lib/storage/image-download-preparation": { scheduleImageDownloadPreparation() {} }, + "@/lib/storage/canvas-cdn-download": { resolveCanvasCdnDownload: async () => ({ task: pending, direct: null, canPrepareDirect: false }) }, + "@/lib/storage/original-image-download": { + prepareOriginalImageArchive: async () => pending, + originalImagePreparing: () => ({ delivery: "preparing", phase: "preparing", requestId: pending.id }), + }, + "@/lib/canvas/download-preparation": { getDownloadPreparationQueue() { throw new Error("Must not relay image bytes"); } }, + })("app/api/canvas/download-image/route.ts"); + const response = await route.POST(new Request("https://app.example/api/canvas/download-image", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ url: pending.sourceUrl, preferDirect: true, preparation: "async" }), + })); + expect(response.status).toBe(202); + expect(await response.json()).toMatchObject({ delivery: "preparing", phase: "preparing" }); +}); diff --git a/tests/routes/original-image-download.test.ts b/tests/routes/original-image-download.test.ts index 25d64a09b..3219ffa85 100644 --- a/tests/routes/original-image-download.test.ts +++ b/tests/routes/original-image-download.test.ts @@ -24,7 +24,7 @@ test("original downloads use the archived JPEG rather than the prepared PNG and expect(new URL(result.downloadUrl).pathname).toBe("/original.jpg"); }); -test("signed original URLs retain the byte-preserving proxy without changing their signature", async () => { +test("signed original URLs cannot silently fall back to the application server", async () => { const module = serverLoader({ "@/lib/db/queries": { getDefaultStorageConfig: async () => config }, "./image-archive-db": { findImageArchiveDownloadTask: async () => null }, @@ -45,16 +45,17 @@ test("existing CDN objects bypass the archive scan and retain the MIME-derived f expect(new URL(result.downloadUrl).pathname).toBe("/source.png"); }); -test("temporary third-party URLs are never checked with HEAD when no archive is ready", async () => { +test("temporary third-party originals enqueue storage without reading their body", async () => { const module = serverLoader({ "@/lib/db/queries": { getDefaultStorageConfig: async () => config }, - "./image-archive-db": { findImageArchiveDownloadTask: async () => null }, + "./image-archive-db": { findImageArchiveDownloadTask: async () => null, enqueueImageArchiveTask: async () => ({ id: "new", status: "pending" }) }, + "./image-archive-worker": { scheduleImageArchiveWorker() {} }, fetch() { throw new Error("The upstream file service does not support HEAD"); }, })("lib/storage/original-image-download.ts"); - expect(await module.resolveOriginalCdnDownload("https://upstream.example/image.png", "image", new AbortController().signal)).toBeNull(); + expect(await module.resolveOriginalCdnDownload("https://upstream.example/image.png", "image", new AbortController().signal)).toMatchObject({ delivery: "preparing" }); }); -test("downloading a pending original prioritizes its existing archive task and keeps the proxy available", async () => { +test("downloading a pending original prioritizes storage and reports preparation", async () => { const preferred: string[] = []; const module = serverLoader({ "@/lib/db/queries": { getDefaultStorageConfig: async () => config }, @@ -62,7 +63,7 @@ test("downloading a pending original prioritizes its existing archive task and k "./image-archive-worker": { scheduleImageArchiveWorker: (_delivery: boolean, id: string) => preferred.push(id) }, fetch() { throw new Error("The original must remain available through the proxy while it is queued"); }, })("lib/storage/original-image-download.ts"); - expect(await module.resolveOriginalCdnDownload("https://upstream.example/queued.png", "image", new AbortController().signal)).toBeNull(); + expect(await module.resolveOriginalCdnDownload("https://upstream.example/queued.png", "image", new AbortController().signal)).toMatchObject({ delivery: "preparing" }); expect(preferred).toEqual(["pending-image"]); }); @@ -151,14 +152,13 @@ test("chat original downloads use a native CDN download without fetching the ima expect(progress.map(event => event.phase)).toEqual(["preparing", "downloading"]); }); -test("original fallback preserves JPEG bytes and the MIME-derived filename", async () => { - const dom = browserMocks(); const bytes = Buffer.from([255, 216, 255, 224, 0, 16, 74, 70, 73, 70, 255, 217]); let calls = 0; - const client = serverLoader({ ...dom, fetch: async () => ++calls === 1 ? Response.json({ delivery: "proxy" }) : new Response(bytes, { headers: { "content-type": "image/jpeg" } }) })("lib/client/canvas-image-download.ts"); - expect(await client.downloadOriginalImage({ url: "https://external.example/image.png", name: "original.png" })).toEqual({ filename: "original.jpg", bytes: bytes.length }); - expect(calls).toBe(2); - const actual = await fetch(dom.links[0].href).then(r=>r.arrayBuffer()); - expect(Buffer.from(actual).equals(bytes)).toBe(true); - URL.revokeObjectURL(dom.links[0].href); +test("a legacy proxy response does not start a slow original download", async () => { + const dom = browserMocks(); let calls = 0; + const client = serverLoader({ ...dom, fetch: async () => { calls++; return Response.json({ delivery: "proxy" }); } })("lib/client/canvas-image-download.ts"); + await expect(client.downloadOriginalImage({ url: "https://external.example/image.png" })).rejects.toThrow("原图尚未准备完成"); + expect(calls).toBe(1); + expect(dom.links).toHaveLength(0); + expect(dom.frames).toHaveLength(0); }); test("one click prepares a compatible PNG when the CDN original is WebP", async () => { @@ -174,7 +174,7 @@ test("one click prepares a compatible PNG when the CDN original is WebP", async expect(init?.method).toBe("POST"); expect(JSON.parse(String(init?.body))).toMatchObject({ url, name: "商品", preparation: "async" }); if (calls.length === 2) { - return Response.json({ delivery: "preparing", retryAfter: 1, message: "正在准备原图" }, { status: 202 }); + return Response.json({ delivery: "preparing", phase: "converting", retryAfter: 1, message: "正在转换为兼容 PNG，完成后自动下载" }, { status: 202 }); } return Response.json({ delivery: "cdn", downloadUrl: "https://cdn.example/prepared.png", filename: "商品.png", bytes: 900 }); } })("lib/client/canvas-image-download.ts"); @@ -190,31 +190,19 @@ test("one click prepares a compatible PNG when the CDN original is WebP", async expect(dom.links).toHaveLength(0); }); -test("one click uses the PNG preparation route without buffering an incompatible proxy response", async () => { - const dom = browserMocks(); - const calls: string[] = []; - const progress: CanvasImageDownloadProgress[] = []; - let cancelled = false; - const bytes = Uint8Array.of(137, 80, 78, 71, 13, 10, 26, 10); - const client = serverLoader({ ...dom, fetch: async (path: string) => { - calls.push(path); - if (calls.length === 1) return Response.json({ delivery: "proxy" }); - if (calls.length === 2) { - return new Response(new ReadableStream({ cancel() { cancelled = true; } }), - { headers: { "content-type": "image/avif" } }); - } - return new Response(bytes, { headers: { "content-type": "image/png", "content-length": String(bytes.length) } }); +test("queued originals show preparation then download directly without a proxy body", async () => { + const dom = browserMocks(); const calls: string[] = []; const progress: CanvasImageDownloadProgress[] = []; + const client = serverLoader({ ...dom, fetch: async (url: string) => { + calls.push(url); + if (calls.length === 1) return Response.json({ delivery: "preparing", retryAfter: 1, message: "正在保存原图，完成后自动下载" }, { status: 202 }); + return Response.json({ delivery: "cdn", downloadUrl: "https://cdn.example/image.png", filename: "image.png", bytes: 900 }); } })("lib/client/canvas-image-download.ts"); - expect(await client.downloadOriginalImage({ - url: "https://external.example/image", name: "商品", onProgress: (event: CanvasImageDownloadProgress) => progress.push(event), - })).toEqual({ filename: "商品.png", bytes: bytes.length }); - expect(cancelled).toBe(true); - expect(calls[2]).toBe("/api/canvas/download-image"); - expect(progress).toContainEqual({ - phase: "downloading", message: "正在下载 PNG...", loadedBytes: bytes.length, totalBytes: bytes.length, percent: 99, - }); - expect(dom.links).toHaveLength(1); - URL.revokeObjectURL(dom.links[0].href); + await client.downloadOriginalImage({ url: "https://external.example/image", onProgress: (event: CanvasImageDownloadProgress) => progress.push(event) }); + expect(calls).toHaveLength(2); + expect(calls.every(url => url.includes("download=resolve"))).toBe(true); + expect(progress.map(p => p.phase)).toEqual(["preparing", "preparing", "downloading"]); + expect(dom.frames[0].src).toBe("https://cdn.example/image.png"); + expect(dom.links).toHaveLength(0); }); test("original downloads reject processed references before network access", async () => {