diff --git a/src/contexts/UploadQueueContext.jsx b/src/contexts/UploadQueueContext.jsx index afb1433..3877bf3 100644 --- a/src/contexts/UploadQueueContext.jsx +++ b/src/contexts/UploadQueueContext.jsx @@ -5,18 +5,29 @@ // batch survives navigating away from the Add Asset page — the floating // UploadProgressToast reads the same state from anywhere in /admin. // -// One POST /admin/assets/batch per addBatch() call — axios onUploadProgress -// only reports aggregate bytes for the whole multipart body, so every job in -// a batch shares one progress number while "uploading" and only splits into -// per-file uploaded/failed once the response comes back. Retries re-submit -// a single file through the same endpoint so they can track their own -// progress independently of whatever batch they originally belonged to. +// Every job — whether it's part of an initial addBatch() or a single retry — +// goes out as its OWN POST /admin/assets/batch request (that endpoint has +// always accepted 1-20 files; sending 1 is a fully supported path). This +// matters beyond just per-file progress: the backend processes a batch +// request's files sequentially inside one request/response cycle and never +// aborts on client disconnect, so a connection drop anywhere during a +// combined multi-file request used to fail EVERY job in that request, even +// ones the server had already finished (S3 object + DB row both created) — +// clicking Retry on those then silently created a duplicate. One request per +// file means a dropped connection can only affect the one job riding it. +// uploadOne() below is shared by both the initial run and retries; runBatch() +// just fans it out over a small worker pool so a big batch doesn't open 20 +// connections at once (MAX_CONCURRENT mirrors garage-anon-proxy's own +// MAX_CONCURRENT_UPLOADS=3 ceiling on the S3 side, so the frontend doesn't +// just relocate the pile-up into tripping that limit instead). import { createContext, useCallback, useContext, useRef, useState } from "react"; import { nanoid } from "nanoid"; import api from "@/utils/api.util"; import { validateAssetFile } from "@/utils/assetUpload.util"; +const MAX_CONCURRENT = 3; + const UploadQueueContext = createContext(null); export function useUploadQueue() { @@ -36,46 +47,63 @@ export function UploadQueueProvider({ children }) { setJobs((prev) => prev.map((j) => (idSet.has(j.id) ? { ...j, ...(typeof patch === "function" ? patch(j) : patch) } : j))); }, []); - // ─── Submit a batch of already-validated files ───────────────────────── - const runBatch = useCallback(async (batchId, batchJobs, meta) => { - const ids = batchJobs.map((j) => j.id); - patchJobs(ids, { status: "uploading", progress: 0 }); - + // ─── Upload exactly one job through its own request ──────────────────── + // Shared by the initial batch run and single-job retries — a dropped + // connection here can only ever fail the one job riding this request, + // never its batch-mates. + const uploadOne = useCallback(async (job) => { const form = new FormData(); - batchJobs.forEach((j) => form.append("files", j.file)); - Object.entries(meta).forEach(([k, v]) => { + form.append("files", job.file); + Object.entries(job.meta).forEach(([k, v]) => { if (v !== undefined && v !== null) form.append(k, v); }); + // Bulk display_name derivation only applies when the caller didn't + // already provide one — each job keeps whatever name it resolves to. + if (!job.meta.display_name) form.append("display_name", baseNameOf(job.name)); try { const { data } = await api.post("/admin/assets/batch", form, { onUploadProgress: (evt) => { const pct = evt.total ? Math.round((evt.loaded / evt.total) * 100) : 0; - patchJobs(ids, { progress: pct }); + patchJobs([job.id], { progress: pct }); }, }); - - const results = data?.data?.results ?? []; - results.forEach((result, idx) => { - const job = batchJobs[idx]; - if (!job) return; - patchJobs([job.id], result.success - ? { status: "uploaded", progress: 100, asset: result.data, error: null } - : { status: "failed", progress: 100, error: result.message || "Upload failed." }); - }); - + const result = data?.data?.results?.[0]; + patchJobs([job.id], result?.success + ? { status: "uploaded", progress: 100, asset: result.data, error: null } + : { status: "failed", progress: 100, error: result?.message || "Upload failed." }); } catch (err) { // No sonner toast() here — the floating widget (both corners // would collide, see UploadProgressToast) already surfaces this - // via each job's failed status. + // via the job's own failed status. const message = err?.response?.data?.message ?? "Upload failed."; - patchJobs(ids, { status: "failed", progress: 100, error: message }); - } finally { - onSettledRef.current.get(batchId)?.(); - onSettledRef.current.delete(batchId); + patchJobs([job.id], { status: "failed", progress: 100, error: message }); } }, [patchJobs]); + // ─── Run a batch through a small fixed-size worker pool ──────────────── + // One POST per file (uploadOne) fanned out over MAX_CONCURRENT workers, + // instead of one request for the whole batch — isolates a dropped + // connection to the single job it was carrying instead of failing every + // job in the batch (which is what used to happen). + const runBatch = useCallback(async (batchId, batchJobs) => { + const ids = batchJobs.map((j) => j.id); + patchJobs(ids, { status: "uploading", progress: 0 }); + + let next = 0; + const worker = async () => { + while (next < batchJobs.length) { + await uploadOne(batchJobs[next++]); + } + }; + await Promise.all( + Array.from({ length: Math.min(MAX_CONCURRENT, batchJobs.length) }, worker) + ); + + onSettledRef.current.get(batchId)?.(); + onSettledRef.current.delete(batchId); + }, [patchJobs, uploadOne]); + // files: File[]; meta: { is_public, storage_provider, createdBy } const addBatch = useCallback((files, meta, { onSettled } = {}) => { const batchId = nanoid(); @@ -103,7 +131,7 @@ export function UploadQueueProvider({ children }) { if (uploadable.length) onSettledRef.current.set(batchId, onSettled); else onSettled(); } - if (uploadable.length) runBatch(batchId, uploadable, meta); + if (uploadable.length) runBatch(batchId, uploadable); return batchId; }, [runBatch]); @@ -117,32 +145,8 @@ export function UploadQueueProvider({ children }) { return { ...j, status: "uploading", progress: 0, error: null }; })); if (!target) return; - - const form = new FormData(); - form.append("files", target.file); - Object.entries(target.meta).forEach(([k, v]) => { - if (v !== undefined && v !== null) form.append(k, v); - }); - // Bulk display_name derivation only applies when >1 file — a lone - // retry should keep whatever name the file originally resolved to. - if (!target.meta.display_name) form.append("display_name", baseNameOf(target.name)); - - try { - const { data } = await api.post("/admin/assets/batch", form, { - onUploadProgress: (evt) => { - const pct = evt.total ? Math.round((evt.loaded / evt.total) * 100) : 0; - patchJobs([jobId], { progress: pct }); - }, - }); - const result = data?.data?.results?.[0]; - patchJobs([jobId], result?.success - ? { status: "uploaded", progress: 100, asset: result.data, error: null } - : { status: "failed", progress: 100, error: result?.message || "Upload failed." }); - } catch (err) { - const message = err?.response?.data?.message ?? "Upload failed."; - patchJobs([jobId], { status: "failed", progress: 100, error: message }); - } - }, [patchJobs]); + await uploadOne(target); + }, [uploadOne]); const removeJob = useCallback((jobId) => { setJobs((prev) => prev.filter((j) => j.id !== jobId || j.status === "uploading"));