From a9aab5e9eaaf543a270e87ab44a915b014685ea1 Mon Sep 17 00:00:00 2001 From: thetapilla Date: Sun, 5 Jul 2026 17:55:09 +0800 Subject: [PATCH 1/3] feat(upload): add multipart uploader - Add a chunked resumable uploader driving /api/fs/multipart with three concurrent ascending chunk requests - Resume interrupted sessions transparently from init's received ranges and fall back to Stream for single-chunk files - Treat flow control (429/409) and transient network errors as retryable, probing session status before declaring failure - Stop sending remaining chunks once the server reports the session completed (rapid upload) - Register the uploader gated by the multipart_enabled public setting --- src/lang/en/settings.json | 3 + src/pages/home/uploads/multipart.ts | 300 ++++++++++++++++++++++++++++ src/pages/home/uploads/uploads.ts | 8 +- 3 files changed, 310 insertions(+), 1 deletion(-) create mode 100644 src/pages/home/uploads/multipart.ts diff --git a/src/lang/en/settings.json b/src/lang/en/settings.json index d11a06e0b..0bf5e73cd 100755 --- a/src/lang/en/settings.json +++ b/src/lang/en/settings.json @@ -64,6 +64,9 @@ "max_index_depth-tips": "max depth of index", "max_server_download_speed": "Max server download speed", "max_server_upload_speed": "Max server upload speed", + "multipart_chunk_size": "Multipart upload chunk size (MB)", + "multipart_chunk_size-tips": "chunk size of multipart upload in MB (1-90), keep it under your CDN's request body limit", + "multipart_enabled": "Enable multipart upload", "non_efs_zip_encoding": "Alternative encoding for ZIP Files", "ocr_api": "Ocr api", "offline_download_task_threads_num": "Offline download task threads num", diff --git a/src/pages/home/uploads/multipart.ts b/src/pages/home/uploads/multipart.ts new file mode 100644 index 000000000..5f68efc5a --- /dev/null +++ b/src/pages/home/uploads/multipart.ts @@ -0,0 +1,300 @@ +import { getSettingNumber, password } from "~/store" +import { Resp } from "~/types" +import { r } from "~/utils" +import { SetUpload, Upload } from "./types" +import { calculateHash } from "./util" +import { StreamUpload } from "./stream" + +type MultipartState = + | "receiving" + | "completed" + | "failed_retriable" + | "failed_permanent" + | "aborted" + +interface MultipartSnapshot { + upload_id: string + state: MultipartState + attempt: number + path: string + size: number + chunk_size: number + total_chunks: number + received: [number, number][] + received_bytes: number + frontier: number + storage_progress: number + error?: string +} + +type SnapResp = Resp +type InitResp = Resp + +// concurrent chunk requests; the server-side window holds 8 chunks, so 3 +// in-flight ascending uploads virtually never hit flow control +const INFLIGHT = 3 +// flow control (429 / server window full) is not failure: the server already +// parks each chunk request for up to ~10s waiting for a slot, so retries here +// only happen when the storage uplink is genuinely slower than the client — +// be patient rather than declaring a large upload dead +const MAX_FLOW_RETRIES = 400 + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) + +export const MultipartUpload: Upload = async ( + uploadPath: string, + file: File, + setUpload: SetUpload, + _asTask = false, // sessions are synchronous pipelines; As-Task does not apply + overwrite = false, + rapid = false, +): Promise => { + // a single-chunk multipart upload costs 3 requests where Stream costs 1, + // and small files pass CDN body limits anyway — silently fall back + const fallbackThreshold = + Math.max(1, getSettingNumber("multipart_chunk_size", 10)) * 1024 * 1024 + if (file.size <= fallbackThreshold) { + return StreamUpload(uploadPath, file, setUpload, false, overwrite, rapid) + } + + const initHeaders: Record = { + "File-Path": encodeURIComponent(uploadPath), + "X-File-Size": file.size, + "Content-Type": file.type || "application/octet-stream", + "Last-Modified": file.lastModified, + Password: password(), + Overwrite: overwrite.toString(), + } + if (rapid) { + setUpload("status", "hashing") + const { md5, sha1, sha256 } = await calculateHash(file, (p) => { + setUpload("progress", p | 0) + }) + initHeaders["X-File-Md5"] = md5 + initHeaders["X-File-Sha1"] = sha1 + initHeaders["X-File-Sha256"] = sha256 + } + + setUpload("status", "uploading") + setUpload("progress", 0) + // init resumes an interrupted session of the same path+size transparently: + // `received` tells us which chunks to skip; a failed_retriable session + // reports nothing received and re-sending from chunk 0 re-fills it. + // init is idempotent thanks to the resume semantics, so a transient network + // failure is safe to retry instead of failing the whole upload. + let initResp: InitResp | undefined + for (let i = 0; ; i++) { + try { + initResp = await r.post("/fs/multipart/init", undefined, { + headers: initHeaders, + }) + break + } catch (e) { + if (i >= 2) throw e + await sleep(1000 * (i + 1)) + } + } + if (initResp!.code !== 200) { + throw new Error(initResp!.message) + } + const session = initResp!.data + const uploadId = session.upload_id + const chunkSize = session.chunk_size + const total = session.total_chunks + const chunkLen = (idx: number) => + idx === total - 1 ? file.size - idx * chunkSize : chunkSize + + if (session.state === "completed") { + // rapid upload finished before we sent a single chunk + setUpload("progress", 100) + return + } + const have = new Set() + for (const [lo, hi] of session.received ?? []) { + for (let i = lo; i <= hi; i++) have.add(i) + } + const missing: number[] = [] + for (let i = 0; i < total; i++) { + if (!have.has(i)) missing.push(i) + } + + let ackedBytes = session.received_bytes ?? 0 + const inflightLoaded: Record = {} + let lastTime = Date.now() + let lastBytes = ackedBytes + const report = () => { + let inflight = 0 + for (const v of Object.values(inflightLoaded)) inflight += v + const done = Math.min(ackedBytes + inflight, file.size) + setUpload("progress", ((done / file.size) * 100) | 0) + const now = Date.now() + if (now - lastTime >= 1000) { + setUpload("speed", ((done - lastBytes) / (now - lastTime)) * 1000) + lastTime = now + lastBytes = done + } + } + + let completedEarly = false // rapid upload finished the session server-side + let fatal: Error | undefined + + const sendChunk = async (idx: number) => { + const blob = file.slice(idx * chunkSize, idx * chunkSize + chunkLen(idx)) + let flowRetries = 0 + for (;;) { + if (completedEarly || fatal) return + let resp: SnapResp | undefined + try { + resp = await r.put("/fs/multipart/chunk", blob, { + headers: { + "X-Upload-Id": uploadId, + "X-Chunk-Index": idx, + "Content-Type": "application/octet-stream", + }, + onUploadProgress: (e: any) => { + inflightLoaded[idx] = e.loaded ?? 0 + report() + }, + }) + } catch (_) { + // network hiccup, or the server fast-rejected without draining the + // body (flow control) and the browser reported it as an error + resp = undefined + } + delete inflightLoaded[idx] + if (resp && resp.code === 200) { + ackedBytes += chunkLen(idx) + if (resp.data?.state === "completed") { + completedEarly = true + } + report() + return + } + if (resp && resp.code !== 429 && resp.code !== 409) { + throw new Error(resp.message) + } + if (!resp) { + // a rapid upload may have completed the session while this chunk was + // mid-flight; the early server response surfaces as a network error — + // ask the session before assuming the network actually failed + let st: SnapResp | undefined + try { + st = await r.get(`/fs/multipart/status?upload_id=${uploadId}`) + } catch (_) { + st = undefined + } + if ( + (st?.code === 200 && st.data.state === "completed") || + st?.code === 404 // completed sessions are reaped after success + ) { + completedEarly = true + return + } + if (st?.code === 200 && st.data.state.startsWith("failed")) { + throw new Error(st.data.error || `upload failed (${st.data.state})`) + } + } + flowRetries++ + if (flowRetries > MAX_FLOW_RETRIES) { + throw new Error(`chunk ${idx} stalled: the storage cannot keep up`) + } + await sleep(resp ? 800 : Math.min(1200 * flowRetries, 4800)) + } + } + + // refill protocol: a failed_retriable session restarts only when chunk 0 + // arrives, so send it alone first — concurrent siblings would race the + // respawn and be rejected + let startFrom = 0 + if (session.state === "failed_retriable" && missing.length > 0) { + await sendChunk(missing[0]) + startFrom = 1 + } + + // ascending order keeps the server window flowing without back-pressure + let nextIdx = startFrom + const worker = async () => { + for (;;) { + if (completedEarly || fatal) return + const i = nextIdx++ + if (i >= missing.length) return + try { + await sendChunk(missing[i]) + } catch (e: any) { + fatal = e instanceof Error ? e : new Error(String(e)) + return + } + } + } + await Promise.all( + Array.from({ length: Math.min(INFLIGHT, missing.length) || 1 }, worker), + ) + if (fatal) { + throw fatal + } + + // every chunk is on the server; the driver is still writing to the storage + setUpload("status", "backending") + setUpload("speed", 0) + let settled = false + const completePromise: Promise = r + .post("/fs/multipart/complete", undefined, { + headers: { "X-Upload-Id": uploadId }, + }) + .catch(() => undefined) // connection may be cut by a CDN; fall back to polling + .finally(() => { + settled = true + }) as Promise + + while (!settled) { + await sleep(2000) + if (settled) break + try { + const st: SnapResp = await r.get( + `/fs/multipart/status?upload_id=${uploadId}`, + ) + if (st.code === 200 && st.data.state === "receiving") { + setUpload("progress", st.data.storage_progress | 0) + } + } catch (_) { + // transient; keep waiting for complete + } + } + const fin = await completePromise + if (fin) { + if (fin.code !== 200) { + throw new Error(fin.message) + } + setUpload("progress", 100) + return + } + // the complete request never came back — poll the session to its end + for (;;) { + await sleep(2000) + let st: SnapResp + try { + st = await r.get(`/fs/multipart/status?upload_id=${uploadId}`) + } catch (_) { + continue + } + if (st.code === 404) { + // completed sessions are removed right after they are reaped; with all + // chunks acknowledged this means the upload finished + setUpload("progress", 100) + return + } + if (st.code !== 200) { + throw new Error(st.message) + } + switch (st.data.state) { + case "completed": + setUpload("progress", 100) + return + case "receiving": + setUpload("progress", st.data.storage_progress | 0) + continue + default: + throw new Error(st.data.error || `upload failed (${st.data.state})`) + } + } +} diff --git a/src/pages/home/uploads/uploads.ts b/src/pages/home/uploads/uploads.ts index 2ddb1f517..e7d351cbe 100644 --- a/src/pages/home/uploads/uploads.ts +++ b/src/pages/home/uploads/uploads.ts @@ -1,7 +1,8 @@ -import { objStore } from "~/store" +import { getSettingBool, objStore } from "~/store" import { FormUpload } from "./form" import { StreamUpload } from "./stream" import { HttpDirectUpload } from "./direct" +import { MultipartUpload } from "./multipart" import { Upload } from "./types" type Uploader = { @@ -29,6 +30,11 @@ const AllUploads: Uploader[] = [ upload: FormUpload, available: () => true, }, + { + name: "Multipart", + upload: MultipartUpload, + available: () => getSettingBool("multipart_enabled"), + }, ] export const getUploads = (): Pick[] => { From f09bc65dbb92912632b147bdfd3839faf6b0b1bf Mon Sep 17 00:00:00 2001 From: thetapilla Date: Sun, 5 Jul 2026 17:55:24 +0800 Subject: [PATCH 2/3] feat(upload): add per-file retry and multipart UX wiring - Add a Retry button on failed rows that reuses the kept File handle, resuming multipart uploads from the breakpoint - Disable "add as task" while the multipart uploader is selected, since sessions are synchronous pipelines --- src/lang/en/home.json | 3 ++- src/pages/home/uploads/Upload.tsx | 42 +++++++++++++++++++++++++++---- 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/src/lang/en/home.json b/src/lang/en/home.json index 104cc2e83..e3251bce3 100644 --- a/src/lang/en/home.json +++ b/src/lang/en/home.json @@ -182,7 +182,8 @@ "success": "Success", "error": "Error", "back": "Back to Upload", - "clear_done": "Clear Done" + "clear_done": "Clear Done", + "retry": "Retry" }, "local_settings": { "aria2_rpc_url": "Aria2 RPC URL", diff --git a/src/pages/home/uploads/Upload.tsx b/src/pages/home/uploads/Upload.tsx index 3d3a5df71..8d4feca06 100644 --- a/src/pages/home/uploads/Upload.tsx +++ b/src/pages/home/uploads/Upload.tsx @@ -28,7 +28,7 @@ import { File2Upload, traverseFileTree } from "./util" import { SelectWrapper } from "~/components" import { getUploads } from "./uploads" -const UploadFile = (props: UploadFileProps) => { +const UploadFile = (props: UploadFileProps & { onRetry?: () => void }) => { const t = useT() return ( { {getFileSize(props.speed)}/s - {getFileSize(props.size)} + + + + + {getFileSize(props.size)} + { } let fileInput!: HTMLInputElement let folderInput!: HTMLInputElement + // keep the File handles around so a failed row can be retried in place + const fileMap = new Map() const handleAddFiles = async (files: File[]) => { if (files.length === 0) return setUploading(true) for (const file of files) { const upload = File2Upload(file) + fileMap.set(upload.path, file) setUploadFiles("uploads", (uploads) => [...uploads, upload]) } for await (const ms of asyncPool(3, files, handleFile)) { @@ -112,6 +127,17 @@ const Upload = () => { // All upload methods are available by default const uploaders = getUploads() const [curUploader, setCurUploader] = createSignal(uploaders[0]) + // multipart sessions are synchronous pipelines with their own progress and + // retry semantics; "add as task" does not apply to them + const asTaskUnsupported = () => curUploader()?.name === "Multipart" + const retryFile = (path: string) => { + const file = fileMap.get(path) + if (!file) return + setUpload(path, "msg", "") + setUpload(path, "progress", 0) + setUpload(path, "speed", 0) + handleFile(file) + } const handleFile = async (file: File) => { const path = file.webkitRelativePath ? file.webkitRelativePath : file.name setUpload(path, "status", "uploading") @@ -124,7 +150,7 @@ const Upload = () => { (key, value) => { setUpload(path, key, value) }, - uploadConfig.asTask, + asTaskUnsupported() ? false : uploadConfig.asTask, uploadConfig.overwrite, uploadConfig.rapid, ) @@ -173,7 +199,12 @@ const Upload = () => { - {(upload) => } + {(upload) => ( + retryFile(upload.path)} + /> + )} } @@ -303,7 +334,8 @@ const Upload = () => { direction={{ "@initial": "column", "@md": "row" }} > { setUploadConfig({ asTask: !uploadConfig.asTask }) }} From 536c827f926cd6a029e61ab2f0cc8362a0e5fefb Mon Sep 17 00:00:00 2001 From: thetapilla Date: Tue, 7 Jul 2026 12:24:14 +0800 Subject: [PATCH 3/3] fix(upload): keep multipart progress monotonic - Report progress and speed from a high-water mark: a chunk rejected by flow control or a network hiccup drops its in-flight bytes from the running sum until it is resent, which made the displayed speed go negative and the progress bar jump backwards - Rejected chunks now show as a progress plateau with speed falling to zero --- src/pages/home/uploads/multipart.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/pages/home/uploads/multipart.ts b/src/pages/home/uploads/multipart.ts index 5f68efc5a..65a7be006 100644 --- a/src/pages/home/uploads/multipart.ts +++ b/src/pages/home/uploads/multipart.ts @@ -120,18 +120,22 @@ export const MultipartUpload: Upload = async ( let ackedBytes = session.received_bytes ?? 0 const inflightLoaded: Record = {} + // a rejected chunk (flow control, network hiccup) drops its in-flight bytes + // from the sum until it is resent; report the high-water mark so progress + // plateaus instead of jumping backwards and speed can never go negative + let peakDone = ackedBytes let lastTime = Date.now() let lastBytes = ackedBytes const report = () => { let inflight = 0 for (const v of Object.values(inflightLoaded)) inflight += v - const done = Math.min(ackedBytes + inflight, file.size) - setUpload("progress", ((done / file.size) * 100) | 0) + peakDone = Math.max(peakDone, Math.min(ackedBytes + inflight, file.size)) + setUpload("progress", ((peakDone / file.size) * 100) | 0) const now = Date.now() if (now - lastTime >= 1000) { - setUpload("speed", ((done - lastBytes) / (now - lastTime)) * 1000) + setUpload("speed", ((peakDone - lastBytes) / (now - lastTime)) * 1000) lastTime = now - lastBytes = done + lastBytes = peakDone } }