diff --git a/CLAUDE.md b/CLAUDE.md index 91c0b55a..36d3d773 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -57,7 +57,7 @@ A **workspace** — called a **lane** in the API, CLI, database and code, which - **Adopted lanes are not destroyable.** A lane with `kind === "adopted"` may never have its worktree reset or removed; "removing" one only drops the dashboard's own record of it. This is enforced in code (`assertDestroyable`'s first check) — do not add a path that bypasses it. - **Removing a lane deletes the lane's RECORD and nothing else — for every kind, managed included.** `POST /api/lanes/:id/remove` stops the run and the stack, clears AgentHub's own runtime bookkeeping under `LANES_ROOT/.state/lane/`, and deletes the row. It calls neither `removeWorktree` nor `removeLaneData`: the working tree, its branch and the lane's database are the USER's data, and a mis-click on a dashboard row may never cost a branch's unpushed code. Deleting those is the user's own `git worktree remove` / `git branch -d` / db drop. Consequently `remove` has no unpushed-commits force-gate (only `reset` does) and no preflight blocker stops it. Do not add a "also delete the worktree" flag back onto this route. (A session's own `claude --worktree` tree is a different thing with its own guarded route — see the session-worktree rule below; a lane's folder is never it.) - **A lane can hold up to ten tmux sessions (slots 0–9; the Workspace groups them into tabs of up to four split panes) — and only slot 0 is the lane's run.** `agenthub-lane-` is slot 0 and the only one `run_id`/`status` ever record; `agenthub-lane--<1-9>` are extra sessions on the same lane (`MAX_PANE_SLOT` in `server/lib/pty-run.js`; the `-[1-9]` run id pattern is repeated in `pty-attach.js`, `websocket.js`, `routes/hooks.js` and the client — widen them together). Anything that must stop a lane's agents (`stop`, `reset`, `remove`) goes through `laneRunIds` so no slot survives it; anything that names "the lane's pane" (`agenthub lanes shell`) means slot 0. A client that wants "a new tab" sends `slot: "auto"` and the SERVER picks the first free slot under the lane lock (`409 ERUNLIVE` when all ten are busy) — never a slot computed from the client's own list, which may already be stale. -- **The File Explorer (`server/lib/file-explorer.js`, `/api/lanes/:id/fs/*`) reuses the read path's guards and never replaces silently.** Every path is `lanePath` + the PARENT `resolveInLane`-confined with the last segment kept as a name (so a symlink is moved/deleted as a link, never followed); never in or under `.git`, never a secret, never the workspace folder itself. Writes run as the folder's account — a member's only through `write-as` (`move`/`copy` modes; `writeAsAsync` streams an upload to its stdin; a batch upload of many small files is ONE `extract` of a tar the server writes itself — folders and regular files only, names it checked, `tar -k`) — and a move/copy never crosses accounts (`sameOwner`, `EFOREIGN`). A taken name is settled by `conflict` (`fail` → 409 with `names`, `skip`, `keep`, `replace` = the old one to the trash first). Delete is a move into the account's trash (member: `~/.local/share/agenthub/trash/`, host: `/trash//`), restorable 7 days. `write-as` is root-installed: a change to it needs `agenthub server install-system` again (`EUPGRADE` until then). A user's HOME is browsed through the same module inside `inHome` (`/api/homes/:who/fs/*`, `server/lib/homes.js` picks the folder): every dot entry at the TOP of a home stays hidden and refused — through a symlink, a download, a zip or as a copy source too — and another user's home is an admin's READ-ONLY, audited view (`EREADONLY`) — except its TRASH, whose restore/purge an admin may run (`TRASH_W` in `server/routes/homes.js`, the owner's trash via `trashKey`, audited with the item id); keep that exception to the trash routes only. A home's Trash is its USER's one recycle bin (`listUserTrash`: home-account store + host store `/trash//`, items tagged `store`). Do not add a delete that skips the trash, an overwrite without `conflict`, or a request path that is not run through `locate`/`folder`. +- **The File Explorer (`server/lib/file-explorer.js`, `/api/lanes/:id/fs/*`) reuses the read path's guards and never replaces silently.** Every path is `lanePath` + the PARENT `resolveInLane`-confined with the last segment kept as a name (so a symlink is moved/deleted as a link, never followed); never in or under `.git`, never a secret, never the workspace folder itself. Writes run as the folder's account — a member's only through `write-as` (`move`/`copy` modes; `writeAsAsync` streams an upload to its stdin; a batch upload of many small files is ONE `extract` of a tar the server writes itself — folders and regular files only, names it checked, `tar -k`; a member's single file goes to a hidden temp name and is `write-as move`d onto its name only once every byte arrived, never written in place) — and a move/copy never crosses accounts (`sameOwner`, `EFOREIGN`). A taken name is settled by `conflict` (`fail` → 409 with `names`, `skip`, `keep`, `replace` = the old one to the trash first). Delete is a move into the account's trash (member: `~/.local/share/agenthub/trash/`, host: `/trash//`), restorable 7 days. `write-as` is root-installed: a change to it needs `agenthub server install-system` again (`EUPGRADE` until then). A user's HOME is browsed through the same module inside `inHome` (`/api/homes/:who/fs/*`, `server/lib/homes.js` picks the folder): every dot entry at the TOP of a home stays hidden and refused — through a symlink, a download, a zip or as a copy source too — and another user's home is an admin's READ-ONLY, audited view (`EREADONLY`) — except its TRASH, whose restore/purge an admin may run (`TRASH_W` in `server/routes/homes.js`, the owner's trash via `trashKey`, audited with the item id); keep that exception to the trash routes only. A home's Trash is its USER's one recycle bin (`listUserTrash`: home-account store + host store `/trash//`, items tagged `store`). **An upload survives a dashboard restart because its bytes are staged first** (`server/lib/upload-staging.js`): parts go to `/upload-staging///` on the HOST, keyed by the caller's user and a browser-made id (never a path), and `?staged=` on `upload`/`upload-batch` writes them through the SAME code and guards — a commit's answer is stored and replayed, never written twice. Do not add a staged write that skips `upload`/`uploadBatch`, nor stage inside a member's home. Do not add a delete that skips the trash, an overwrite without `conflict`, or a request path that is not run through `locate`/`folder`. - **The whole-file editor and the Files tab's upload are the two non-git writes on the lane file surface, and both live outside `git-ops.js` on purpose.** `git-ops.js` only touches paths `git status` reports as changed; `lane-file-write.js` may save any existing text file, so it must reuse the READ path's guards (`lanePath`, `resolveInLane`, `isSecretOnDisk`) and never be weaker than them, and it saves only over the `baseSha` it was opened at — do not let it create files or touch a binary. `lane-file-transfer.js`'s upload (`POST /api/lanes/:id/git/upload`, issue #78) is the opposite and only that: it CREATES one file (any bytes, ≤ 25 MB) under the same guards, into an existing folder, refuses a taken name (`EEXISTS`, a symlink included) and anything into or named `.git` (a hook there runs on the dashboard's next git call), and writes as the folder's member through `write-as` under S3 — do not let it overwrite, make folders or take a path from anything but `dir` + one plain name. Its `GET …/git/download` twin streams any file through `openSafe` as a `sandbox`ed attachment. The tree, expand-all and quick-open read the disk (`walkFolder`) for a folder that is not a repo. - **Server tests never touch the user's tmux.** Run them through `npm run test:server` (`scripts/test-server.js`), never a bare `node --test server/__tests__/*.test.js` from inside a lane pane — the suites create and kill real `agenthub-lane-` sessions. - **An artifact is shared as a SNAPSHOT and installed only into the viewer's own account.** `server/lib/artifacts.js` (`/api/artifacts`): a pinned file or a skill is copied once into `/artifacts//` and every viewer reads only that copy (an owner's edit of a skill — in the Hub, or in the skill's source folder, which the server takes when the Hub is listed and an installer's differing copy is then offered **Update** — writes a NEW snapshot and drops the old one — never a write into the one being read; the newer of the two changes wins; **Share again** (`POST /api/artifacts/:id/reshare`) takes the owner's folder on the server — else a user-scope plugin's same-named skill — on a click, and an upload named like the uploader's own skill, or carrying `id`, replaces that row's snapshot instead of adding a twin) — `raw` serves only a file the snapshot recorded, under a `sandbox` CSP + `nosniff`; a `page` is a `https://claude.ai/…` link only. A developer sees their own rows plus shared ones; only the owner or an admin changes one. Install writes a skill into BOTH `~/.claude/skills/` and the cross-tool `~/.agents/skills/` of every account the CALLER's sessions run as (`installAccounts`: their S3 member home, plus the host's for an admin, whose host-folder workspaces run as the host; S3: as them through `write-as`), never over an existing one without `replace` (the old one is moved to `~/.claude/agenthub-replaced-skills/`, never deleted), and runs a plugin's `claude plugin marketplace add`/`install --scope user` as the caller, never with `-y`; Uninstall (`POST /api/artifacts/:id/uninstall`) takes it out of the same accounts — a skill MOVED to `agenthub-replaced-skills/`, never deleted, a plugin `claude plugin uninstall --scope user --keep-data`; an admin's switch (Settings → AI tools, `ArtifactsHubSettings`) turns installs off. An uploaded or written skill is snapshotted under the same checks (paths relative, no `..`, no secret names, zip sizes read before inflating). `POST /api/artifacts/plugin/inspect` clones a user-named https link only to read `marketplace.json`, with every credential helper OFF — never send a login with it. Do not add a route that serves a live file of another user, installs into someone else's account, or takes a plugin command from the request. diff --git a/client/src/components/BackgroundJobsDock.tsx b/client/src/components/BackgroundJobsDock.tsx index e2a323e3..88deadd3 100644 --- a/client/src/components/BackgroundJobsDock.tsx +++ b/client/src/components/BackgroundJobsDock.tsx @@ -287,6 +287,11 @@ function UploadChip({ job }: { job: UploadJob }) { {isArmed("stop") && !stopping && (

{t("uploads.stopConfirm")}

)} + {job.waiting && !stopping && ( +

+ {t("uploads.waiting")} +

+ )}

{job.folder} {job.name ? ` · ${job.name}` : ""} diff --git a/client/src/components/lanes/FileExplorer.tsx b/client/src/components/lanes/FileExplorer.tsx index ffdd3e27..24f6a42a 100644 --- a/client/src/components/lanes/FileExplorer.tsx +++ b/client/src/components/lanes/FileExplorer.tsx @@ -91,6 +91,7 @@ import { type FsPlace, type GitSource, isHomePlace, + newUploadId, } from "../../lib/api"; import type { Lane } from "../../lib/types"; import { useDialog } from "../../hooks/useDialog"; @@ -688,13 +689,21 @@ export function FileBrowser({ // false once the server says a batch cannot be written here (an old helper). let batchable = true; const show = (name: string, extra = 0) => job.update({ name, done, sent: sent + extra }); + const onWait = (waiting: boolean) => job.update({ waiting }); /** One file; false = the user cancelled. */ const one = async ({ file, rel }: { file: File; rel: string }) => { if (signal.aborted) return false; const dest = join(dir, rel); show(rel); + // One id for every attempt: a retry with another `conflict` reuses the staged bytes. + const upload = newUploadId(); const send = (c: FsConflict) => - api.files.upload(place, dest, file, c, (n) => show(rel, n), signal); + api.files.uploadStaged(place, dest, file, c, { + upload, + onProgress: (n) => show(rel, n), + onWait, + signal, + }); try { const r = await send(forAll ?? "fail").catch(async (e) => { if (errCode(e) !== "EEXISTS") throw e; @@ -707,6 +716,9 @@ export function FileBrowser({ } catch (e) { if (errCode(e) === "ECANCEL" || errCode(e) === "EABORT") return false; toast.error(t("fx.uploadFailed", { name: rel, message: errText(e) })); + } finally { + onWait(false); + void api.files.dropUpload(place, upload); } sent += file.size; done++; @@ -726,6 +738,7 @@ export function FileBrowser({ return a.choice; }; if (signal.aborted) return "cancel"; + const upload = newUploadId(); try { const c = await api.files.uploadCheck( place, @@ -739,7 +752,12 @@ export function FileBrowser({ let policy = await choose(c.taken); if (!policy) return "cancel"; const send = (p: FsConflict) => - api.files.uploadBatch(place, dir, b, p, (n) => show(label, n), signal); + api.files.uploadBatchStaged(place, dir, b, p, { + upload, + onProgress: (n) => show(label, n), + onWait, + signal, + }); const r = await send(policy).catch(async (e) => { // Taken meanwhile: ask about those, then once more. if (errCode(e) !== "EEXISTS") throw e; @@ -758,6 +776,9 @@ export function FileBrowser({ return "fallback"; } toast.error(t("fx.uploadFailed", { name: label, message: errText(e) })); + } finally { + onWait(false); + void api.files.dropUpload(place, upload); } sent += size; done += b.length; diff --git a/client/src/components/lanes/__tests__/FileExplorer.test.tsx b/client/src/components/lanes/__tests__/FileExplorer.test.tsx index 7c94649a..1dfaac7d 100644 --- a/client/src/components/lanes/__tests__/FileExplorer.test.tsx +++ b/client/src/components/lanes/__tests__/FileExplorer.test.tsx @@ -31,6 +31,10 @@ vi.mock("../../../lib/api", () => ({ gitTarget: (lane: number, session?: string | null, source?: number | null) => source ? { lane, source } : session ? { lane, session } : lane, isHomePlace: (p: unknown) => typeof p === "object" && p !== null && "home" in p, + newUploadId: (() => { + let n = 0; + return () => `upload-id-${String(++n).padStart(8, "0")}`; + })(), api: { lanes: { list: vi.fn(), @@ -52,8 +56,9 @@ vi.mock("../../../lib/api", () => ({ allTrash: vi.fn(), restore: vi.fn(), purge: vi.fn(), - upload: vi.fn(), - uploadBatch: vi.fn(), + uploadStaged: vi.fn(), + uploadBatchStaged: vi.fn(), + dropUpload: vi.fn(async () => null), uploadCheck: vi.fn(), zipUrl: (t: number, ps: string[]) => `/api/lanes/${t}/fs/zip?p=${ps.join("&p=")}`, }, @@ -236,7 +241,7 @@ describe("FileExplorer", () => { }); it("asks about a taken name on upload, and sends the choice", async () => { - vi.mocked(api.files.upload) + vi.mocked(api.files.uploadStaged) .mockRejectedValueOnce( Object.assign(new Error("taken"), { code: "EEXISTS", names: ["a.txt"] }) ) @@ -249,20 +254,24 @@ describe("FileExplorer", () => { const ask = await screen.findByTestId("fx-conflict"); await user.click(within(ask).getByRole("button", { name: "Keep both" })); await waitFor(() => - expect(api.files.upload).toHaveBeenLastCalledWith( + expect(api.files.uploadStaged).toHaveBeenLastCalledWith( 1, "a.txt", file, "keep", - expect.any(Function), - expect.any(AbortSignal) + expect.objectContaining({ upload: expect.any(String), signal: expect.any(AbortSignal) }) ) ); + // The retry names the SAME upload, so the staged bytes are not sent again, + // and the staged upload is dropped once answered. + const [first, second] = vi.mocked(api.files.uploadStaged).mock.calls.map((c) => c[4].upload); + expect(second).toBe(first); + await waitFor(() => expect(api.files.dropUpload).toHaveBeenCalledWith(1, first)); }); it("sends many small files as one batch, asking once about every taken name", async () => { vi.mocked(api.files.uploadCheck).mockResolvedValue({ taken: ["a.txt", "b.txt"], batch: true }); - vi.mocked(api.files.uploadBatch).mockResolvedValue({ saved: 3, skipped: 0, paths: [] }); + vi.mocked(api.files.uploadBatchStaged).mockResolvedValue({ saved: 3, skipped: 0, paths: [] }); const user = userEvent.setup(); open(); await screen.findByText("a.txt"); @@ -271,33 +280,32 @@ describe("FileExplorer", () => { const ask = await screen.findByTestId("fx-conflict"); await user.click(within(ask).getByRole("button", { name: "Replace" })); await waitFor(() => - expect(api.files.uploadBatch).toHaveBeenCalledWith( + expect(api.files.uploadBatchStaged).toHaveBeenCalledWith( 1, "", files.map((file) => ({ file, rel: file.name })), "replace", - expect.any(Function), - expect.any(AbortSignal) + expect.objectContaining({ upload: expect.any(String), signal: expect.any(AbortSignal) }) ) ); expect(api.files.uploadCheck).toHaveBeenCalledWith(1, "", ["a.txt", "b.txt", "c.txt"]); - expect(api.files.upload).not.toHaveBeenCalled(); + expect(api.files.uploadStaged).not.toHaveBeenCalled(); }); it("falls back to one request per file when the server cannot take a batch", async () => { vi.mocked(api.files.uploadCheck).mockResolvedValue({ taken: [], batch: false }); - vi.mocked(api.files.upload).mockResolvedValue({ path: "x" }); + vi.mocked(api.files.uploadStaged).mockResolvedValue({ path: "x" }); const user = userEvent.setup(); open(); await screen.findByText("a.txt"); const files = ["x.txt", "y.txt"].map((n) => new File(["hi"], n)); await user.upload(screen.getByTestId("fx-upload-files"), files); - await waitFor(() => expect(api.files.upload).toHaveBeenCalledTimes(2)); - expect(api.files.uploadBatch).not.toHaveBeenCalled(); + await waitFor(() => expect(api.files.uploadStaged).toHaveBeenCalledTimes(2)); + expect(api.files.uploadBatchStaged).not.toHaveBeenCalled(); }); it("leaves a picked repo's .git out of a folder upload, with one notice", async () => { - vi.mocked(api.files.upload).mockResolvedValue({ path: "x" }); + vi.mocked(api.files.uploadStaged).mockResolvedValue({ path: "x" }); open(); await screen.findByText("a.txt"); const mk = (rel: string) => { @@ -307,8 +315,8 @@ describe("FileExplorer", () => { }; const files = ["repo/.git/HEAD", "repo/.git/objects/d4/ab", "repo/src/a.js"].map(mk); fireEvent.change(screen.getByTestId("fx-upload-folder"), { target: { files } }); - await waitFor(() => expect(api.files.upload).toHaveBeenCalledTimes(1)); - expect(vi.mocked(api.files.upload).mock.calls[0]![1]).toBe("repo/src/a.js"); + await waitFor(() => expect(api.files.uploadStaged).toHaveBeenCalledTimes(1)); + expect(vi.mocked(api.files.uploadStaged).mock.calls[0]![1]).toBe("repo/src/a.js"); expect(toast.info).toHaveBeenCalledTimes(1); expect(toast.error).not.toHaveBeenCalled(); }); @@ -320,7 +328,7 @@ describe("FileExplorer", () => { Object.defineProperty(big, "size", { value: 600 * 1024 * 1024 }); fireEvent.change(screen.getByTestId("fx-upload-files"), { target: { files: [big] } }); await waitFor(() => expect(toast.error).toHaveBeenCalled()); - expect(api.files.upload).not.toHaveBeenCalled(); + expect(api.files.uploadStaged).not.toHaveBeenCalled(); }); it("restores from the Trash", async () => { diff --git a/client/src/i18n/locales/en/lanes.json b/client/src/i18n/locales/en/lanes.json index 97427c24..a9dfd603 100644 --- a/client/src/i18n/locales/en/lanes.json +++ b/client/src/i18n/locales/en/lanes.json @@ -262,6 +262,7 @@ "jobs.expand": "Show background tasks", "uploads.stop": "Stop upload", "uploads.stopConfirm": "Click again to stop", + "uploads.waiting": "The dashboard is restarting — the upload resumes when it is back.", "uploads.stopping": "Stopping…", "uploads.stopped_one": "Upload stopped — {{count}} of {{total}} file sent.", "uploads.stopped_other": "Upload stopped — {{count}} of {{total}} files sent.", diff --git a/client/src/i18n/locales/vi/lanes.json b/client/src/i18n/locales/vi/lanes.json index 495b4b7e..71a0ff23 100644 --- a/client/src/i18n/locales/vi/lanes.json +++ b/client/src/i18n/locales/vi/lanes.json @@ -262,6 +262,7 @@ "jobs.expand": "Xem các việc chạy nền", "uploads.stop": "Ngưng tải lên", "uploads.stopConfirm": "Bấm lần nữa để ngưng", + "uploads.waiting": "Dashboard đang khởi động lại — tải lên sẽ tiếp tục khi có kết nối.", "uploads.stopping": "Đang ngưng…", "uploads.stopped_one": "Đã ngưng tải lên — đã gửi {{count}}/{{total}} tệp.", "uploads.stopped_other": "Đã ngưng tải lên — đã gửi {{count}}/{{total}} tệp.", diff --git a/client/src/lib/__tests__/uploadStaged.test.ts b/client/src/lib/__tests__/uploadStaged.test.ts new file mode 100644 index 00000000..99751ed5 --- /dev/null +++ b/client/src/lib/__tests__/uploadStaged.test.ts @@ -0,0 +1,134 @@ +/** + * @file uploadStaged.test.ts + * @description The resumable File Explorer upload (`api.files.uploadStaged`): + * a file goes in parts to `/fs/upload-part` and is committed with + * `?staged=`; a dashboard that goes away mid-way (network error, a + * proxy's bare 502) is waited out and the upload resumes from the byte the + * server reports; a refusal with our own error code is not retried. + * @author Nguyễn Ngọc Trí Vĩ + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { api, newUploadId } from "../api"; + +type Answer = { status: number; body?: unknown } | "neterr"; + +/** A fake XHR that answers each send with the next scripted answer. */ +class FakeXhr { + static answers: ((url: string, size: number) => Answer)[] = []; + static sent: { url: string; size: number }[] = []; + status = 0; + responseText = ""; + upload: { onprogress: ((e: { loaded: number }) => void) | null } = { onprogress: null }; + onload: (() => void) | null = null; + onerror: (() => void) | null = null; + onabort: (() => void) | null = null; + private url = ""; + open(_m: string, url: string) { + this.url = url; + } + setRequestHeader() {} + abort() { + this.onabort?.(); + } + send(body: Blob) { + FakeXhr.sent.push({ url: this.url, size: body.size }); + const next = FakeXhr.answers.shift(); + const a = next ? next(this.url, body.size) : { status: 500, body: { error: { code: "EX" } } }; + queueMicrotask(() => { + if (a === "neterr") return this.onerror?.(); + this.status = a.status; + this.responseText = a.body === undefined ? "" : JSON.stringify(a.body); + this.onload?.(); + }); + } +} + +const MB = 1024 * 1024; +const q = (url: string) => new URL(url, "http://x").searchParams; + +beforeEach(() => { + vi.useFakeTimers(); + FakeXhr.answers = []; + FakeXhr.sent = []; + vi.stubGlobal("XMLHttpRequest", FakeXhr); +}); +afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); +}); + +describe("api.files.uploadStaged", () => { + it("sends parts, waits out a restart, resumes from the server's count, then commits", async () => { + const file = new Blob([new Uint8Array(20 * MB)]); + const upload = newUploadId(); + expect(upload).toMatch(/^[0-9a-f]{32}$/); + // part 1 ok; part 2 dies with the server; part 2 again after the restart; + // part 3; the commit's answer is lost once; then the stored answer. + FakeXhr.answers = [ + () => ({ status: 200, body: { received: 8 * MB } }), + () => "neterr", + (url) => ({ status: 200, body: { received: Number(q(url).get("offset")) + 8 * MB } }), + () => ({ status: 200, body: { received: 20 * MB } }), + () => ({ status: 502 }), + () => ({ status: 201, body: { path: "big.bin" } }), + ]; + // While away: the status read says 10 MB arrived before the server died. + const fetchMock = vi.fn( + async () => + new Response(JSON.stringify({ received: 10 * MB, total: 20 * MB, done: false }), { + headers: { "content-type": "application/json" }, + }) + ); + vi.stubGlobal("fetch", fetchMock); + const waits: boolean[] = []; + const done = api.files.uploadStaged(7, "big.bin", file, "fail", { + upload, + onWait: (w) => waits.push(w), + }); + await vi.runAllTimersAsync(); + await expect(done).resolves.toEqual({ path: "big.bin" }); + const parts = FakeXhr.sent.filter((s) => s.url.includes("/upload-part")); + expect(parts.map((p) => [q(p.url).get("offset"), p.size])).toEqual([ + ["0", 8 * MB], + [String(8 * MB), 8 * MB], + [String(10 * MB), 8 * MB], + [String(18 * MB), 2 * MB], + ]); + for (const p of parts) expect(q(p.url).get("upload")).toBe(upload); + const commits = FakeXhr.sent.filter((s) => s.url.includes("/fs/upload?")); + expect(commits).toHaveLength(2); + expect(q(commits[0]!.url).get("staged")).toBe(upload); + expect(q(commits[0]!.url).get("path")).toBe("big.bin"); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(waits).toEqual([true, false, true, false]); + }); + + it("goes on from the server's count on EOFFSET, and never retries our own refusal", async () => { + const file = new Blob([new Uint8Array(10)]); + FakeXhr.answers = [ + () => ({ status: 409, body: { error: { code: "EOFFSET", message: "x", received: 10 } } }), + () => ({ status: 409, body: { error: { code: "EEXISTS", message: "taken", names: ["a"] } } }), + ]; + const done = api.files.uploadStaged(7, "a", file, "fail", { upload: newUploadId() }); + const settled = expect(done).rejects.toMatchObject({ code: "EEXISTS", names: ["a"] }); + await vi.runAllTimersAsync(); + await settled; + expect(FakeXhr.sent).toHaveLength(2); + }); + + it("gives up once the dashboard has been away too long", async () => { + const file = new Blob([new Uint8Array(4)]); + FakeXhr.answers = Array.from({ length: 200 }, () => () => "neterr" as const); + vi.stubGlobal( + "fetch", + vi.fn(async () => { + throw new TypeError("Failed to fetch"); + }) + ); + const done = api.files.uploadStaged(7, "a", file, "fail", { upload: newUploadId() }); + const settled = expect(done).rejects.toMatchObject({ code: "ENET" }); + await vi.runAllTimersAsync(); + await settled; + }); +}); diff --git a/client/src/lib/api.ts b/client/src/lib/api.ts index a438fc34..e593c03d 100644 --- a/client/src/lib/api.ts +++ b/client/src/lib/api.ts @@ -1188,7 +1188,9 @@ const sendRaw = ( xhr.upload.onprogress = (e) => onProgress?.(e.loaded); xhr.onerror = () => reject(Object.assign(new Error("network error"), { code: "ENET" })); xhr.onload = () => { - let body: { error?: { code?: string; message?: string; names?: string[] } } = {}; + let body: { + error?: { code?: string; message?: string; names?: string[]; received?: number }; + } = {}; try { body = JSON.parse(xhr.responseText); } catch { @@ -1206,12 +1208,148 @@ const sendRaw = ( code: body.error?.code, status: xhr.status, names: body.error?.names, + received: body.error?.received, }) ); }; xhr.send(payload); }); +/** + * A batch body (`/fs/upload-batch`): a 4-byte length, the JSON file list, + * then each file's bytes. The Blob is only composed, never read into memory. + */ +function batchBody(files: { file: Blob; rel: string }[]): Blob { + const head = new TextEncoder().encode( + JSON.stringify({ + files: files.map((f) => ({ + path: f.rel, + size: f.file.size, + mtime: f.file instanceof File ? f.file.lastModified : undefined, + })), + }) + ); + const len = new Uint8Array(4); + new DataView(len.buffer).setUint32(0, head.length); + return new Blob([len, head, ...files.map((f) => f.file)], { type: "application/octet-stream" }); +} + +/** One part of a resumable upload (the server takes at most 16 MB). */ +const UPLOAD_PART_BYTES = 8 * 1024 * 1024; +/** How long an upload waits for the dashboard to come back (a restart, an update). */ +const UPLOAD_WAIT_MS = 3 * 60 * 1000; + +/** A fresh id for one resumable upload (also outside a secure context). */ +export function newUploadId(): string { + const b = new Uint8Array(16); + crypto.getRandomValues(b); + return Array.from(b, (x) => x.toString(16).padStart(2, "0")).join(""); +} + +type SendErr = Error & { code?: string; status?: number; received?: number }; + +/** + * The dashboard is away (restarting) rather than refusing: worth waiting for. + * A 5xx with none of our error codes is a proxy (nginx, Vite) answering for it. + */ +const isAway = (e: SendErr) => + e.code === "ENET" || e.code === "EBUSY" || ((e.status ?? 0) >= 500 && !e.code); + +const pause = (ms: number, signal?: AbortSignal) => + new Promise((resolve, reject) => { + const timer = setTimeout(resolve, ms); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timer); + reject(Object.assign(new Error("aborted"), { code: "EABORT" })); + }, + { once: true } + ); + }); + +/** Callbacks of a resumable upload. */ +export interface StagedSendOpts { + /** The upload's id: pass the SAME one to retry a commit (another `conflict`) without resending. */ + upload: string; + onProgress?: (sent: number) => void; + /** true while the dashboard is away and the upload waits to resume. */ + onWait?: (waiting: boolean) => void; + signal?: AbortSignal; +} + +/** + * Send `payload` in parts to `place`'s staging (`/fs/upload-part`), then + * POST `commitUrl(id)` to write it. A dropped connection or a restarting + * dashboard (network error, 502/503/504) is waited out for up to + * UPLOAD_WAIT_MS, and the upload resumes from the byte the server has. + */ +async function sendStaged( + place: FsPlace, + commitUrl: (id: string) => string, + payload: Blob, + { upload, onProgress, onWait, signal }: StagedSendOpts +): Promise { + const total = payload.size; + let received = 0; + let awaySince = 0; + let delay = 500; + const waitOut = async (e: SendErr) => { + if (!isAway(e)) throw e; + if (!awaySince) awaySince = Date.now(); + if (Date.now() - awaySince > UPLOAD_WAIT_MS) throw e; + onWait?.(true); + await pause(delay, signal); + delay = Math.min(delay * 2, 5000); + }; + const back = () => { + if (awaySince) onWait?.(false); + awaySince = 0; + delay = 500; + }; + const partUrl = (q: Record) => + fsUrl(place, "/upload-part", new URLSearchParams({ upload, ...q })); + // An empty file is still one (empty) part: it makes the upload known. + do { + try { + const r = await sendRaw<{ received: number }>( + partUrl({ offset: String(received), total: String(total) }), + payload.slice(received, received + UPLOAD_PART_BYTES), + (n) => onProgress?.(received + n), + signal + ); + received = r.received; + back(); + } catch (err) { + const e = err as SendErr; + if (e.code === "EOFFSET" && typeof e.received === "number") { + // The server has more (or less) than we thought: go on from there. + received = e.received; + back(); + continue; + } + await waitOut(e); + try { + const s = await request<{ received: number; done: boolean }>(partUrl({})); + received = s.received; + } catch { + /* still away: the next part asks again */ + } + } + onProgress?.(received); + } while (received < total); + for (;;) { + try { + const r = await sendRaw(commitUrl(upload), new Blob([]), undefined, signal); + back(); + return r; + } catch (err) { + // A commit whose answer was lost is answered again, never written twice. + await waitOut(err as SendErr); + } + } +} + /** A home the File Explorer can open (`GET /api/homes`): `me`, or another user's for an admin. */ export interface FsHome { id: string; @@ -3300,6 +3438,28 @@ export const api = { onProgress, signal ), + /** + * {@link files.upload}, resumable: sent in parts and written once whole, + * so a dashboard restart mid-way only pauses it ({@link sendStaged}). + */ + uploadStaged: ( + t: FsPlace, + path: string, + file: Blob, + conflict: FsConflict, + opts: StagedSendOpts + ) => + sendStaged<{ path: string; skipped?: boolean }>( + t, + (staged) => fsUrl(t, "/upload", new URLSearchParams({ path, conflict, staged })), + file, + opts + ), + /** Forget a resumable upload's staged bytes (stopped, or given up). Never throws. */ + dropUpload: (t: FsPlace, upload: string) => + request<{ dropped: boolean }>(fsUrl(t, "/upload-part", new URLSearchParams({ upload })), { + method: "DELETE", + }).catch(() => null), /** * Many small files in ONE request, under the folder `dir` (each `rel` * relative to it): a 4-byte length, the JSON file list, then the bytes. @@ -3312,25 +3472,27 @@ export const api = { conflict: FsConflict = "fail", onProgress?: (sent: number) => void, signal?: AbortSignal - ) => { - const head = new TextEncoder().encode( - JSON.stringify({ - files: files.map((f) => ({ - path: f.rel, - size: f.file.size, - mtime: f.file instanceof File ? f.file.lastModified : undefined, - })), - }) - ); - const len = new Uint8Array(4); - new DataView(len.buffer).setUint32(0, head.length); - return sendRaw<{ saved: number; skipped: number; paths: string[] }>( + ) => + sendRaw<{ saved: number; skipped: number; paths: string[] }>( fsUrl(t, "/upload-batch", new URLSearchParams({ dir, conflict })), - new Blob([len, head, ...files.map((f) => f.file)], { type: "application/octet-stream" }), + batchBody(files), onProgress, signal - ); - }, + ), + /** {@link files.uploadBatch}, resumable like {@link files.uploadStaged}. */ + uploadBatchStaged: ( + t: FsPlace, + dir: string, + files: { file: Blob; rel: string }[], + conflict: FsConflict, + opts: StagedSendOpts + ) => + sendStaged<{ saved: number; skipped: number; paths: string[] }>( + t, + (staged) => fsUrl(t, "/upload-batch", new URLSearchParams({ dir, conflict, staged })), + batchBody(files), + opts + ), /** Before a batch: which of `paths` (under `dir`) are taken, and whether a batch can go there. */ uploadCheck: (t: FsPlace, dir: string, paths: string[]) => request<{ taken: string[]; batch: boolean }>(fsUrl(t, "/upload-check"), { diff --git a/client/src/lib/uploadJobs.ts b/client/src/lib/uploadJobs.ts index c771c264..cc858f1b 100644 --- a/client/src/lib/uploadJobs.ts +++ b/client/src/lib/uploadJobs.ts @@ -9,6 +9,8 @@ * kept in localStorage (refreshed every ~2 s), and the next page load shows * a record that stopped being refreshed as an interrupted upload until it is * dismissed. A fresh record belongs to another open tab and is left alone. + * A dashboard RESTART is not an interruption: the upload is sent in parts + * (`api.files.uploadStaged`) and the job shows `waiting` until it resumes. * @author Nguyễn Ngọc Trí Vĩ */ @@ -25,6 +27,8 @@ export interface UploadJob { sent: number; bytes: number; state: "running" | "stopping" | "interrupted"; + /** The dashboard is away (a restart): the upload waits and then resumes. */ + waiting?: boolean; } type Stored = UploadJob & { beat: number }; @@ -115,7 +119,8 @@ export function beginUpload(init: { folder: string; total: number; bytes: number return { id, signal: ctrl.signal, - update: (patch: Partial>) => set(id, patch), + update: (patch: Partial>) => + set(id, patch), end: () => { controllers.delete(id); uploads = uploads.filter((u) => u.id !== id); diff --git a/docs/API.md b/docs/API.md index a5e582b6..a9fc0f33 100644 --- a/docs/API.md +++ b/docs/API.md @@ -290,6 +290,9 @@ POST /api/lanes/:id/fs/trash/purge { id | "all" } POST /api/lanes/:id/fs/upload?path=&conflict= raw body, streamed → 201 { path } | 200 { skipped: true } POST /api/lanes/:id/fs/upload-check { dir?, paths } → { taken: [paths], batch } POST /api/lanes/:id/fs/upload-batch?dir=&conflict= length + JSON list + bytes → 201 { saved, skipped, paths } +POST /api/lanes/:id/fs/upload-part?upload=&offset=&total= raw bytes (≤ 16 MB) → { received, total } +GET /api/lanes/:id/fs/upload-part?upload= → { received, total, done } +DELETE /api/lanes/:id/fs/upload-part?upload= → { dropped } GET /api/lanes/:id/fs/zip?p=&p=… application/zip, streamed ``` @@ -328,6 +331,33 @@ install-system` is re-run, and the page then falls back to one request per file. Measured on loopback (host folder): 1,000 files of 2 KB took 4.3 s one by one, 0.5 s as one batch. +The Files page sends every upload — a file or a batch — **resumably** +(`server/lib/upload-staging.js`), so a dashboard restart (an update, a manual +`systemctl restart`, a crash) only pauses it. The browser names the upload +with a random id (`upload`, 16–64 of `[A-Za-z0-9_-]`) and sends its bytes in +8 MB parts to `upload-part`, which appends each at `offset` to +`/upload-staging///` on the HOST (never in a member's +home) — `409 EOFFSET` with `error.received` when a part does not start where +the staged bytes end, `409 EBUSY` while another part of the same id is still +arriving, `413 ETOOBIG` past `total` (512 MB at most), `429 ETOOMANY` beyond 8 +unfinished uploads per user. Then `upload?…&staged=` or +`upload-batch?…&staged=` writes the staged bytes through the SAME code +and guards as a body would (`409 EINCOMPLETE` before every byte is there); a +commit already made answers its stored answer again (a restart may have +swallowed the first), and a refused one (`EEXISTS`) keeps the bytes for a +retry with another `conflict`. On a network error or a bare 5xx (a proxy +answering for a restarting dashboard) the browser waits up to 3 minutes, +asks `GET upload-part` how much arrived and sends the rest; the dock shows +"The dashboard is restarting…" meanwhile. Uploads are per user — another +user's id is never found — and one untouched for a day is swept. A reload +of the page still loses an upload: the bytes are the browser's. + +A file uploaded into a member's folder is written to a hidden +`..agenthub-upload-` by `write-as` and moved onto its name only +once every byte arrived, so a dropped connection or a dying server never +leaves a truncated file under the real name (a `write-as` older than `move` +keeps the old `replace`). + #### File Explorer on a home ``` @@ -353,6 +383,9 @@ POST /api/homes/:who/fs/trash/purge POST /api/homes/:who/fs/upload?path=&conflict= POST /api/homes/:who/fs/upload-check POST /api/homes/:who/fs/upload-batch?dir=&conflict= +POST /api/homes/:who/fs/upload-part?upload=&offset=&total= +GET /api/homes/:who/fs/upload-part?upload= +DELETE /api/homes/:who/fs/upload-part?upload= ``` The same File Explorer on a user's HOME (`server/lib/homes.js`): under S3 their diff --git a/server/__tests__/file-explorer.test.js b/server/__tests__/file-explorer.test.js index a0b81b95..01ac0fe1 100644 --- a/server/__tests__/file-explorer.test.js +++ b/server/__tests__/file-explorer.test.js @@ -4,7 +4,9 @@ * place, across workspaces), delete into the trash and restore/purge, a * streamed upload into new sub-folders with every conflict policy, a batch * of many small files in one request (and, in a member's folder, as ONE - * write-as `extract` of a tar that real `tar` unpacks), a zip of + * write-as `extract` of a tar that real `tar` unpacks), a resumable upload + * sent in parts and committed from the staged bytes (and a member's upload + * written to a hidden temp file, moved onto its name only when whole), a zip of * files and folders that unzips back to the same bytes — and the guards: no * `..`, no symlink walking out, nothing in `.git`, no secret file, the * workspace folder itself never moved or deleted, cross-origin refused. @@ -293,6 +295,144 @@ describe("File Explorer", () => { assert.ok(!fs.existsSync(path.join(W1, "up/s.txt"))); }); + it("resumes an upload sent in parts, and writes it only once every byte is staged", async () => { + const staging = require("../lib/upload-staging"); + const id = "resume-test-0000000001"; + const data = Buffer.from("0123456789abcdefghij"); + const part = (offset, bytes, upload = id) => + request( + "POST", + `/api/lanes/${lane1}/fs/upload-part?upload=${upload}&offset=${offset}&total=${data.length}`, + bytes, + { "Content-Type": "application/octet-stream" } + ); + const status = async () => + (await request("GET", `/api/lanes/${lane1}/fs/upload-part?upload=${id}`)).body; + assert.deepEqual(await status(), { received: 0, total: null, done: false }); + assert.equal((await part(0, data.subarray(0, 8))).body.received, 8); + // A part that does not start where the staged bytes end: 409 with the truth. + const off = await part(4, data.subarray(4, 12)); + assert.equal(off.status, 409); + assert.equal(off.body.error.code, "EOFFSET"); + assert.equal(off.body.error.received, 8); + // Committing early writes nothing. + const early = await request( + "POST", + `/api/lanes/${lane1}/fs/upload?path=res/file.bin&staged=${id}` + ); + assert.equal(early.status, 409); + assert.equal(early.body.error.code, "EINCOMPLETE"); + assert.ok(!fs.existsSync(path.join(W1, "res/file.bin"))); + // A part past the end is refused. + assert.equal((await part(8, Buffer.alloc(13))).status, 413); + assert.equal((await status()).received, 8); + assert.equal((await part(8, data.subarray(8))).body.received, data.length); + const commit = () => + request("POST", `/api/lanes/${lane1}/fs/upload?path=res/file.bin&staged=${id}`); + const done = await commit(); + assert.equal(done.status, 201); + assert.deepEqual(done.body, { path: "res/file.bin" }); + assert.deepEqual(fs.readFileSync(path.join(W1, "res/file.bin")), data); + // A commit whose answer was lost is answered again, never written twice. + fs.writeFileSync(path.join(W1, "res/file.bin"), "changed since"); + const again = await commit(); + assert.equal(again.status, 201); + assert.deepEqual(again.body, { path: "res/file.bin" }); + assert.equal(fs.readFileSync(path.join(W1, "res/file.bin"), "utf8"), "changed since"); + assert.deepEqual(await status(), { received: data.length, total: data.length, done: true }); + // A taken name keeps the staged bytes for a retry with another policy. + const id2 = "resume-test-0000000002"; + await part(0, data, id2); + const taken = await request( + "POST", + `/api/lanes/${lane1}/fs/upload?path=res/file.bin&staged=${id2}` + ); + assert.equal(taken.status, 409); + assert.equal(taken.body.error.code, "EEXISTS"); + const kept = await request( + "POST", + `/api/lanes/${lane1}/fs/upload?path=res/file.bin&staged=${id2}&conflict=keep` + ); + assert.equal(kept.body.path, "res/file (1).bin"); + assert.deepEqual(fs.readFileSync(path.join(W1, "res/file (1).bin")), data); + // A batch goes the same way. + const head = Buffer.from(JSON.stringify({ files: [{ path: "b1.txt", size: 2 }] })); + const len = Buffer.alloc(4); + len.writeUInt32BE(head.length); + const batch = Buffer.concat([len, head, Buffer.from("hi")]); + const id3 = "resume-test-0000000003"; + await request( + "POST", + `/api/lanes/${lane1}/fs/upload-part?upload=${id3}&offset=0&total=${batch.length}`, + batch, + { "Content-Type": "application/octet-stream" } + ); + const b = await request("POST", `/api/lanes/${lane1}/fs/upload-batch?dir=res&staged=${id3}`); + assert.equal(b.status, 201); + assert.equal(fs.readFileSync(path.join(W1, "res/b1.txt"), "utf8"), "hi"); + // Stop: forgotten. A bad id or an unknown upload is refused, never a path. + assert.equal( + (await request("DELETE", `/api/lanes/${lane1}/fs/upload-part?upload=${id}`)).status, + 200 + ); + assert.equal((await status()).received, 0); + assert.equal( + (await request("GET", `/api/lanes/${lane1}/fs/upload-part?upload=../../etc`)).status, + 400 + ); + assert.equal( + (await request("POST", `/api/lanes/${lane1}/fs/upload?path=x&staged=nope-nope-nope-nope`)) + .status, + 404 + ); + // Uploads are per user: another user's key never sees this one. + assert.equal(staging.status("42", id3).received, 0); + assert.ok(fs.statSync(path.join(staging.root(), "local")).isDirectory()); + for (const i of [id2, id3]) staging.drop("local", i); + }); + + it("keeps the bytes of a part cut short, caps unfinished uploads, sweeps old ones", async () => { + const staging = require("../lib/upload-staging"); + const { Readable } = require("node:stream"); + const id = "cut-short-0000000001"; + let sent = false; + const cut = new Readable({ + read() { + if (!sent) { + sent = true; + this.push(Buffer.from("abcd")); + } else setTimeout(() => this.destroy(new Error("connection reset")), 30); + }, + }); + await assert.rejects(staging.putPart("u1", id, { offset: 0, total: 10, stream: cut })); + assert.equal(staging.status("u1", id).received, 4); + const rest = Readable.from([Buffer.from("efghij")]); + assert.equal( + (await staging.putPart("u1", id, { offset: 4, total: 10, stream: rest })).received, + 10 + ); + // At most MAX_OPEN unfinished uploads per user. + for (let i = 1; i < staging.MAX_OPEN; i++) { + await staging.putPart("u1", `cap-test-${String(i).padStart(10, "0")}`, { + offset: 0, + total: 5, + stream: Readable.from([Buffer.from("a")]), + }); + } + await assert.rejects( + staging.putPart("u1", "cap-test-overflow-01", { + offset: 0, + total: 5, + stream: Readable.from([Buffer.from("a")]), + }), + { code: "ETOOMANY" } + ); + // A day untouched: swept. + staging.sweep(Date.now() + staging.KEEP_MS + 60_000); + assert.equal(staging.status("u1", id).received, 0); + assert.deepEqual(fs.readdirSync(path.join(staging.root(), "u1")), []); + }); + it("zips files and folders into one download that unzips to the same bytes", async () => { const r = await request("GET", `/api/lanes/${lane1}/fs/zip?p=src&p=site/css/app.css`); assert.equal(r.status, 200); @@ -404,6 +544,49 @@ describe("File Explorer in a member's folder (S3, run-as stubbed)", () => { } }); + it("writes an upload to a hidden temp file and moves it onto the name only when complete", async () => { + const { Readable } = require("node:stream"); + const prev = runAs.writeAsAsync; + const runs = []; + // write-as as a shell would run it: `cat >file` takes whatever arrived. + runAs.writeAsAsync = async (_a, file, args, { input } = {}) => { + runs.push([path.basename(file), ...args.map((a) => path.basename(a))]); + if (!args.length) { + const chunks = []; + for await (const c of input) chunks.push(c); + fs.writeFileSync(file, Buffer.concat(chunks)); + } else if (args[0] === "move") fs.renameSync(file, args[1]); + else if (args[0] === "remove") fs.rmSync(file, { force: true }); + }; + try { + const out = await fx.upload(M, "up.txt", Readable.from([Buffer.from("whole")]), { + length: "5", + }); + assert.equal(out.path, "up.txt"); + assert.equal(fs.readFileSync(path.join(M, "up.txt"), "utf8"), "whole"); + assert.match(runs[0][0], /^\.up\.txt\.agenthub-upload-/); + assert.deepEqual(runs[1].slice(1), ["move", "up.txt"]); + // The body ended early (a dropped connection, a dying server): nothing + // under the real name, and the temp file is removed. + runs.length = 0; + await assert.rejects( + fx.upload(M, "cut.txt", Readable.from([Buffer.from("par")]), { length: "10" }), + { code: "EBADBODY" } + ); + assert.ok(!fs.existsSync(path.join(M, "cut.txt"))); + assert.deepEqual( + fs.readdirSync(M).filter((n) => n.includes("agenthub-upload")), + [] + ); + assert.deepEqual( + runs.map((r) => r[1] ?? "write"), + ["write", "remove"] + ); + } finally { + runAs.writeAsAsync = prev; + } + }); + it("says the helper needs updating instead of failing obscurely", async () => { runAs.writeAsCanMove = () => false; try { diff --git a/server/__tests__/home-files.test.js b/server/__tests__/home-files.test.js index 2a15a532..0f2f9d56 100644 --- a/server/__tests__/home-files.test.js +++ b/server/__tests__/home-files.test.js @@ -6,6 +6,7 @@ * fine, delete/restore work with no workspace, a copy goes home ↔ workspace * under each side's own rule — and in server mode a developer never reaches * another user's home while an admin gets it read-only — except its Trash, + * and a resumable upload in parts lands in the home under the same rule, * which the admin lists, restores from and purges as the member's own. * @author Nguyễn Ngọc Trí Vĩ */ @@ -166,6 +167,38 @@ describe("File Explorer on the home", () => { assert.ok(!fs.existsSync(path.join(WS, "id_ed25519"))); }); + it("resumes an upload into the home in parts, under the home rule", async () => { + const raw = (method, p, body) => + fetch(`${base}${p}`, { + method, + headers: { "Content-Type": "application/octet-stream" }, + body, + }).then(async (r) => ({ status: r.status, body: await r.json() })); + const id = "home-resume-00000001"; + const data = Buffer.from("home bytes"); + assert.equal( + ( + await raw( + "POST", + `${H}/upload-part?upload=${id}&offset=0&total=${data.length}`, + data.subarray(0, 4) + ) + ).body.received, + 4 + ); + assert.equal((await request("GET", `${H}/upload-part?upload=${id}`)).body.received, 4); + await raw( + "POST", + `${H}/upload-part?upload=${id}&offset=4&total=${data.length}`, + data.subarray(4) + ); + // The home rule still holds for the staged bytes: no top-level dot entry. + assert.equal((await raw("POST", `${H}/upload?path=.ssh/x&staged=${id}`)).status, 403); + const ok = await raw("POST", `${H}/upload?path=docs/up.txt&staged=${id}`); + assert.equal(ok.status, 201); + assert.equal(fs.readFileSync(path.join(HOME, "docs/up.txt"), "utf8"), "home bytes"); + }); + it("refuses an unknown home", async () => { assert.equal((await request("GET", "/api/homes/someone/fs/list")).status, 404); }); @@ -256,13 +289,10 @@ describe("an admin manages a member's Trash", () => { // …and every user's in one list, each item tagged with its home and owner. const all = (await call("GET", "/api/homes/trash", dan)).body; - assert.deepEqual( - all.items.map((i) => [i.name, i.home, i.owner]).sort(), - [ - ["junk", "cara", "cara"], - ["old", "cara", "cara"], - ] - ); + assert.deepEqual(all.items.map((i) => [i.name, i.home, i.owner]).sort(), [ + ["junk", "cara", "cara"], + ["old", "cara", "cara"], + ]); // A developer's merged view is only their own. assert.deepEqual( (await call("GET", "/api/homes/trash", cara)).body.items.map((i) => i.home), diff --git a/server/lib/file-explorer.js b/server/lib/file-explorer.js index 873f3f24..9df6dbf7 100644 --- a/server/lib/file-explorer.js +++ b/server/lib/file-explorer.js @@ -634,16 +634,26 @@ async function purgeTrash(dir, userKey, id) { // ── Upload ───────────────────────────────────────────────────────────────── -/** A pass-through that fails once more than `max` bytes went by. */ +/** A pass-through that fails once more than `max` bytes went by (`seen()`: the count). */ function capped(max) { let n = 0; - return new Transform({ + const t = new Transform({ transform(chunk, _enc, cb) { n += chunk.length; if (n > max) return cb(bad("ETOOBIG", `a file may be at most ${max / 1024 / 1024} MB`)); cb(null, chunk); }, }); + t.seen = () => n; + return t; +} + +/** A body that stopped short of its announced length (a dropped connection). */ +function shortBody(limit, length) { + if (length == null || length === "") return null; + const want = Number(length); + if (!Number.isSafeInteger(want) || limit.seen() === want) return null; + return bad("EBADBODY", `the upload ended early (${limit.seen()} of ${want} bytes)`); } /** @@ -677,16 +687,41 @@ async function upload(dir, rel, stream, { conflict, length, trash } = {}) { } const acct = accountAt(loc.abs); const limit = capped(UPLOAD_MAX_BYTES); - if (acct) { - // write-as `replace`: a new file renamed over the name, as the member. - const done = runAs.writeAsAsync(acct, loc.abs, ["replace"], { input: limit }); - await Promise.all([ + const feed = (done) => + Promise.all([ pipeline(stream, limit).catch((e) => { limit.destroy(e); throw e; }), done, ]); + if (acct && runAs.writeAsCanMove()) { + // As the member: a hidden temp file, moved onto the name only once every + // byte arrived. write-as cannot tell a dropped connection (or a server + // that died mid-upload) from the end of the file — writing the name + // itself would leave a truncated file there under its real name. + const tmp = path.join( + loc.parent.abs, + `.${loc.name}.agenthub-upload-${crypto.randomBytes(4).toString("hex")}` + ); + try { + await feed(runAs.writeAsAsync(acct, tmp, [], { input: limit })); + const short = shortBody(limit, length); + if (short) throw short; + // `move` never lands on an existing name: one taken meanwhile is refused. + await runAs.writeAsAsync(acct, tmp, ["move", loc.abs]).catch((e) => { + if (/exists/.test(e.message)) { + throw bad("EEXISTS", `${loc.rel} already exists`, { names: [loc.name] }); + } + throw e; + }); + } catch (e) { + await runAs.writeAsAsync(acct, tmp, ["remove"]).catch(() => {}); + throw e; + } + } else if (acct) { + // An older write-as without `move`: `replace` (a new file renamed over the name). + await feed(runAs.writeAsAsync(acct, loc.abs, ["replace"], { input: limit })); } else { const tmp = path.join( loc.parent.abs, @@ -694,6 +729,8 @@ async function upload(dir, rel, stream, { conflict, length, trash } = {}) { ); try { await pipeline(stream, limit, fs.createWriteStream(tmp, { flags: "wx" })); + const short = shortBody(limit, length); + if (short) throw short; // Linked, not renamed: a name taken meanwhile is refused, never replaced. await fs.promises.link(tmp, loc.abs).catch(async (e) => { if (e.code === "EEXIST") diff --git a/server/lib/upload-staging.js b/server/lib/upload-staging.js new file mode 100644 index 00000000..d13409b9 --- /dev/null +++ b/server/lib/upload-staging.js @@ -0,0 +1,252 @@ +/** + * @file Resumable uploads for the File Explorer: an upload's bytes are sent in + * parts and kept on the HOST's disk under + * `/upload-staging///` (0700/0600; never in a member's + * home), so a restart of the dashboard — an update, a manual restart, a crash — + * only costs the part in flight: the browser asks how much arrived (`status`) + * and sends the rest. Only when every byte is here is the file written where + * it goes, by the SAME `upload`/`uploadBatch` code (and guards) as a one-shot + * upload, reading the staged file instead of the request (`commit`). + * + * `` is the browser's own random name for one upload; ids live per user + * key, so one user can never read, extend or commit another's. A part must + * start exactly where the staged bytes end (`EOFFSET` with `received`), one + * request per id at a time (`EBUSY`), never past the `total` announced first. + * A committed upload keeps its answer (`result.json`) and drops its bytes, so + * a commit whose answer a restart swallowed is answered again, never written + * twice; a failed commit (a taken name, say) keeps the bytes for a retry with + * another `conflict`. At most `MAX_OPEN` unfinished uploads per user; anything + * untouched for `KEEP_MS` is swept. + * @author Nguyễn Ngọc Trí Vĩ + */ + +"use strict"; + +const fs = require("node:fs"); +const path = require("node:path"); +const { Transform } = require("node:stream"); +const { pipeline } = require("node:stream/promises"); + +const { getDataDir } = require("./claude-home"); + +/** The largest upload (the browser sends a whole file — or a batch — under one id). */ +const TOTAL_MAX = 512 * 1024 * 1024; +/** The largest part one request may carry. */ +const PART_MAX = 16 * 1024 * 1024; +const MAX_OPEN = 8; +const KEEP_MS = 24 * 60 * 60 * 1000; + +const ID_RE = /^[A-Za-z0-9_-]{16,64}$/; +const KEY_RE = /^[A-Za-z0-9_-]{1,64}$/; + +const bad = (code, message, extra = {}) => Object.assign(new Error(message), { code, ...extra }); + +/** Ids (`/`) a request in THIS process is working on. */ +const busy = new Set(); + +const root = () => path.join(getDataDir(), "upload-staging"); + +function dirOf(userKey, id) { + const key = String(userKey ?? ""); + if (!KEY_RE.test(key)) throw bad("EBADBODY", "no user for this upload"); + if (typeof id !== "string" || !ID_RE.test(id)) { + throw bad("EBADBODY", "an upload id is 16–64 letters, digits, '-' or '_'"); + } + return { key, id, dir: path.join(root(), key, id), lock: `${key}/${id}` }; +} + +const dataFile = (d) => path.join(d.dir, "data"); +const metaFile = (d) => path.join(d.dir, "meta.json"); +const resultFile = (d) => path.join(d.dir, "result.json"); + +function readJson(file) { + try { + return JSON.parse(fs.readFileSync(file, "utf8")); + } catch { + return null; + } +} + +function writeJson(file, value) { + const tmp = `${file}.tmp`; + fs.writeFileSync(tmp, JSON.stringify(value), { mode: 0o600 }); + fs.renameSync(tmp, file); +} + +const sizeOf = (d) => { + try { + return fs.statSync(dataFile(d)).size; + } catch { + return 0; + } +}; + +function int(v, name, max) { + const n = typeof v === "string" && /^\d+$/.test(v) ? Number(v) : v; + if (!Number.isSafeInteger(n) || n < 0 || n > max) throw bad("EBADBODY", `bad ${name}`); + return n; +} + +/** Drop uploads untouched for KEEP_MS (every user's — the store is one). */ +function sweep(now = Date.now()) { + let keys = []; + try { + keys = fs.readdirSync(root()); + } catch { + return; + } + for (const key of keys) { + let ids = []; + try { + ids = fs.readdirSync(path.join(root(), key)); + } catch { + continue; + } + for (const id of ids) { + if (busy.has(`${key}/${id}`)) continue; + const dir = path.join(root(), key, id); + try { + if (now - fs.statSync(dir).mtimeMs > KEEP_MS) + fs.rmSync(dir, { recursive: true, force: true }); + } catch { + /* gone meanwhile */ + } + } + } +} + +/** Unfinished uploads of `key` (a committed one keeps only its answer). */ +function openCount(key) { + let ids = []; + try { + ids = fs.readdirSync(path.join(root(), key)); + } catch { + return 0; + } + return ids.filter((id) => !fs.existsSync(path.join(root(), key, id, "result.json"))).length; +} + +/** Touch the folder so `sweep` counts from the last part, not the first. */ +function touch(d) { + const now = new Date(); + try { + fs.utimesSync(d.dir, now, now); + } catch { + /* best effort */ + } +} + +/** `{received, total, done}` of one upload (`received: 0`, `total: null` when unknown). */ +function status(userKey, id) { + const d = dirOf(userKey, id); + const meta = readJson(metaFile(d)); + if (!meta) return { received: 0, total: null, done: false }; + const done = fs.existsSync(resultFile(d)); + return { received: done ? meta.total : sizeOf(d), total: meta.total, done }; +} + +/** + * Append the request `stream` at `offset` of upload `id` (of `total` bytes in + * all). Returns `{received, total}`. + */ +async function putPart(userKey, id, { offset, total, stream }) { + const d = dirOf(userKey, id); + const at = int(offset, "offset", TOTAL_MAX); + const all = int(total, "total", TOTAL_MAX); + if (busy.has(d.lock)) throw bad("EBUSY", "this upload is already receiving a part"); + busy.add(d.lock); + try { + let meta = readJson(metaFile(d)); + if (!meta) { + sweep(); + if (openCount(d.key) >= MAX_OPEN) { + throw bad("ETOOMANY", `at most ${MAX_OPEN} unfinished uploads at a time`); + } + fs.mkdirSync(d.dir, { recursive: true, mode: 0o700 }); + meta = { total: all, createdAt: new Date().toISOString() }; + writeJson(metaFile(d), meta); + } + if (meta.total !== all) throw bad("EBADBODY", `this upload is ${meta.total} bytes`); + if (fs.existsSync(resultFile(d))) return { received: all, total: all }; + const have = sizeOf(d); + if (at !== have) { + throw bad("EOFFSET", `this upload has ${have} bytes, not ${at}`, { received: have }); + } + const room = Math.min(PART_MAX, all - have); + let n = 0; + const cap = new Transform({ + transform(chunk, _enc, cb) { + n += chunk.length; + if (n > room) return cb(bad("ETOOBIG", "this part goes past the upload's end")); + cb(null, chunk); + }, + }); + // A part cut short leaves what arrived: the next one starts from there. + await pipeline(stream, cap, fs.createWriteStream(dataFile(d), { flags: "a", mode: 0o600 })); + touch(d); + return { received: sizeOf(d), total: all }; + } finally { + busy.delete(d.lock); + } +} + +/** + * Write a fully staged upload: `write(stream, size)` is the one-shot upload's + * own code, given the staged bytes; it answers `{status, body}`. A commit + * already made answers its stored answer again; a failed one keeps the bytes. + */ +async function commit(userKey, id, write) { + const d = dirOf(userKey, id); + const meta = readJson(metaFile(d)); + if (!meta) throw bad("ENOFILE", "no such upload — send it again"); + if (busy.has(d.lock)) throw bad("EBUSY", "this upload is still being written"); + busy.add(d.lock); + try { + const stored = readJson(resultFile(d)); + if (stored) return stored; + const have = sizeOf(d); + if (have !== meta.total) { + throw bad("EINCOMPLETE", `this upload has ${have} of ${meta.total} bytes`, { + received: have, + }); + } + const stream = fs.createReadStream(dataFile(d)); + let out; + try { + out = await write(stream, have); + } finally { + stream.destroy(); + } + writeJson(resultFile(d), out); + fs.rmSync(dataFile(d), { force: true }); + touch(d); + return out; + } finally { + busy.delete(d.lock); + } +} + +/** + * Forget one upload (the browser stopped it, or has its answer). Also while a + * part is still arriving: a Stop drops the request, and the part writes on + * into a file no longer listed — never refused, or the stopped upload would + * hold one of MAX_OPEN for a day. + */ +function drop(userKey, id) { + const d = dirOf(userKey, id); + fs.rmSync(d.dir, { recursive: true, force: true }); + return { dropped: true }; +} + +module.exports = { + status, + putPart, + commit, + drop, + sweep, + root, + TOTAL_MAX, + PART_MAX, + MAX_OPEN, + KEEP_MS, +}; diff --git a/server/openapi-extra/files.js b/server/openapi-extra/files.js index d7777f34..ad0307f8 100644 --- a/server/openapi-extra/files.js +++ b/server/openapi-extra/files.js @@ -21,7 +21,7 @@ const tags = [ ]; const errorCodes = - "400 EBADPATH/EBADNAME, 403 ESECRET/EFOREIGN/EREADONLY (another user's home)/EBADORIGIN, 404 ENOFILE/ENOHOME, 409 EEXISTS (with `error.names`)/EUPGRADE/EMEMBERGONE, 413 ETOOBIG."; + "400 EBADPATH/EBADNAME, 403 ESECRET/EFOREIGN/EREADONLY (another user's home)/EBADORIGIN, 404 ENOFILE/ENOHOME, 409 EEXISTS (with `error.names`)/EUPGRADE/EMEMBERGONE, 413 ETOOBIG; a resumable upload adds 409 EOFFSET/EINCOMPLETE (with `error.received`)/EBUSY and 429 ETOOMANY."; const schemas = { FsEntry: { @@ -141,12 +141,109 @@ const OPS = [ parameters: [ { name: "path", in: "query", required: true, schema: { type: "string" } }, { name: "conflict", in: "query", schema: conflict }, + { + name: "staged", + in: "query", + schema: { type: "string" }, + description: + "Write the bytes sent before through `/upload-part` under this upload id instead of a body. A commit made before answers its stored answer again; a failed one keeps the bytes for a retry.", + }, + ], + body: { + required: false, + content: { "application/octet-stream": { schema: { type: "string", format: "binary" } } }, + }, + status: 201, + }, + ], + [ + "get", + "/upload-part", + "How much of a resumable upload arrived", + { + opId: "UploadPartStatus", + note: "`{received, total, done}` for the caller's own upload id (`received: 0, total: null` when unknown). The browser asks this after the dashboard came back, then sends the rest.", + parameters: [ + { + name: "upload", + in: "query", + required: true, + schema: { type: "string", pattern: "^[A-Za-z0-9_-]{16,64}$" }, + }, + ], + ok: { + description: "`{received, total, done}`", + content: { + "application/json": { + schema: { + type: "object", + properties: { + received: { type: "integer" }, + total: { type: "integer", nullable: true }, + done: { type: "boolean" }, + }, + }, + }, + }, + }, + }, + ], + [ + "post", + "/upload-part", + "Send one part of a resumable upload (≤ 16 MB a part, 512 MB in all)", + { + opId: "UploadPart", + note: "Raw bytes appended at `offset` to the caller's upload `upload` of `total` bytes, staged on the host (`/upload-staging/`) until `/upload` or `/upload-batch` names it with `?staged=`. A part must start where the staged bytes end (409 EOFFSET with `error.received`), one at a time per id (409 EBUSY); a part cut short keeps what arrived. At most 8 unfinished uploads per user (429 ETOOMANY); one untouched for a day is swept.", + parameters: [ + { + name: "upload", + in: "query", + required: true, + schema: { type: "string", pattern: "^[A-Za-z0-9_-]{16,64}$" }, + }, + { name: "offset", in: "query", required: true, schema: { type: "integer", minimum: 0 } }, + { name: "total", in: "query", required: true, schema: { type: "integer", minimum: 0 } }, ], body: { required: true, content: { "application/octet-stream": { schema: { type: "string", format: "binary" } } }, }, - status: 201, + ok: { + description: "`{received, total}`", + content: { + "application/json": { + schema: { + type: "object", + properties: { received: { type: "integer" }, total: { type: "integer" } }, + }, + }, + }, + }, + }, + ], + [ + "delete", + "/upload-part", + "Forget a resumable upload's staged bytes", + { + opId: "UploadPartDrop", + parameters: [ + { + name: "upload", + in: "query", + required: true, + schema: { type: "string", pattern: "^[A-Za-z0-9_-]{16,64}$" }, + }, + ], + ok: { + description: "`{dropped: true}`", + content: { + "application/json": { + schema: { type: "object", properties: { dropped: { type: "boolean" } } }, + }, + }, + }, }, ], [ @@ -178,9 +275,16 @@ const OPS = [ parameters: [ { name: "dir", in: "query", schema: { type: "string", default: "" } }, { name: "conflict", in: "query", schema: conflict }, + { + name: "staged", + in: "query", + schema: { type: "string" }, + description: + "Write the bytes sent before through `/upload-part` under this upload id instead of a body. A commit made before answers its stored answer again; a failed one keeps the bytes for a retry.", + }, ], body: { - required: true, + required: false, content: { "application/octet-stream": { schema: { type: "string", format: "binary" } } }, }, status: 201, @@ -233,7 +337,7 @@ function operations(prefix, idParam, opPrefix, note) { tags: ["Files"], summary, description: `${note} ${x.note ? `${x.note} ` : ""}${errorCodes}`, - operationId: `${opPrefix}${sub.replace(/[/-](\w)/g, (_, c) => c.toUpperCase())}`, + operationId: `${opPrefix}${x.opId || sub.replace(/[/-](\w)/g, (_, c) => c.toUpperCase())}`, parameters: [idParam, ...(x.parameters || [])], ...(x.body ? { requestBody: x.body } : {}), responses: { [x.status || 200]: x.ok || { description: "Done." } }, diff --git a/server/routes/homes.js b/server/routes/homes.js index 964e03d1..67573f28 100644 --- a/server/routes/homes.js +++ b/server/routes/homes.js @@ -8,6 +8,7 @@ * GET /api/homes/:who/fs/list|trash|zip|download * POST /api/homes/:who/fs/mkdir|newfile|rename|move|copy|delete|upload|trash/restore|trash/purge * POST /api/homes/:who/fs/upload-batch|upload-check many small files in one request + * GET|POST|DELETE /api/homes/:who/fs/upload-part a resumable upload's parts (then `?staged=`) * * `:who` is `me`, or — for an admin — another user's name: read-only * (`403 EREADONLY` on every write) and written to the audit log. The one @@ -24,6 +25,7 @@ const express = require("express"); const path = require("node:path"); const fx = require("../lib/file-explorer"); +const staging = require("../lib/upload-staging"); const homes = require("../lib/homes"); const scope = require("../lib/scope"); const lanesLib = require("../lib/lanes"); @@ -46,12 +48,17 @@ function fail(res, err) { EUPGRADE: 409, EMEMBERGONE: 409, ETOOBIG: 413, + EOFFSET: 409, + EBUSY: 409, + EINCOMPLETE: 409, + ETOOMANY: 429, }[err.code] ?? 500; res.status(status).json({ error: { code: err.code || "EFS", message: err.message, ...(Array.isArray(err.names) ? { names: err.names } : {}), + ...(Number.isInteger(err.received) ? { received: err.received } : {}), }, }); } @@ -294,22 +301,57 @@ router.post( TRASH_W ) ); +/** + * The body of an upload: the request, or with `?staged=` the bytes sent + * before in parts (`upload-staging.js`; see the workspace routes). + */ +function stagedOr(req, write) { + if (req.query.staged === undefined) return write(req, req.headers["content-length"]); + req.resume(); + return staging.commit(userKey(req), req.query.staged, write); +} +router.get( + "/:who/fs/upload-part", + handle(async (req, res) => res.json(staging.status(userKey(req), req.query.upload)), W) +); +router.post( + "/:who/fs/upload-part", + sameOriginGuard, + handle(async (req, res) => { + try { + res.json( + await staging.putPart(userKey(req), req.query.upload, { + offset: req.query.offset, + total: req.query.total, + stream: req, + }) + ); + } catch (err) { + req.resume(); + throw err; + } + }, W) +); +router.delete( + "/:who/fs/upload-part", + sameOriginGuard, + handle(async (req, res) => res.json(staging.drop(userKey(req), req.query.upload)), W) +); router.post( "/:who/fs/upload", sameOriginGuard, handle(async (req, res, home) => { try { - const out = await fx.upload( - home.dir, - typeof req.query.path === "string" ? req.query.path : "", - req, - { - conflict: req.query.conflict, - length: req.headers["content-length"], - trash: trashFor(req, home), - } - ); - res.status(out.skipped ? 200 : 201).json(out); + const r = await stagedOr(req, async (stream, length) => { + const out = await fx.upload( + home.dir, + typeof req.query.path === "string" ? req.query.path : "", + stream, + { conflict: req.query.conflict, length, trash: trashFor(req, home) } + ); + return { status: out.skipped ? 200 : 201, body: out }; + }); + res.status(r.status).json(r.body); } catch (err) { req.resume(); throw err; @@ -321,13 +363,16 @@ router.post( sameOriginGuard, handle(async (req, res, home) => { try { - const out = await fx.uploadBatch( - home.dir, - typeof req.query.dir === "string" ? req.query.dir : "", - req, - { conflict: req.query.conflict, trash: trashFor(req, home) } - ); - res.status(201).json(out); + const r = await stagedOr(req, async (stream) => ({ + status: 201, + body: await fx.uploadBatch( + home.dir, + typeof req.query.dir === "string" ? req.query.dir : "", + stream, + { conflict: req.query.conflict, trash: trashFor(req, home) } + ), + })); + res.status(r.status).json(r.body); } catch (err) { req.resume(); throw err; diff --git a/server/routes/lanes.js b/server/routes/lanes.js index 244e586c..55c76a69 100644 --- a/server/routes/lanes.js +++ b/server/routes/lanes.js @@ -1879,6 +1879,7 @@ router.post("/:id/git/upload", sameOriginGuard, uploadBody, async (req, res) => // folder's account. `conflict` (fail|skip|keep|replace) settles a taken // name; 409 EEXISTS carries `names` so the browser can ask. const fileExplorer = require("../lib/file-explorer"); +const uploadStaging = require("../lib/upload-staging"); function fsFail(res, err) { const status = @@ -1894,12 +1895,17 @@ function fsFail(res, err) { EMEMBERGONE: 409, ETOOBIG: 413, ENOHOME: 404, + EOFFSET: 409, + EBUSY: 409, + EINCOMPLETE: 409, + ETOOMANY: 429, }[err.code] ?? 500; res.status(status).json({ error: { code: err.code || "EFS", message: err.message, ...(Array.isArray(err.names) ? { names: err.names } : {}), + ...(Number.isInteger(err.received) ? { received: err.received } : {}), }, }); } @@ -2056,6 +2062,61 @@ router.post("/:id/fs/trash/purge", sameOriginGuard, async (req, res) => { } }); +/** + * An upload's body: the request itself, or — with `?staged=` — the bytes + * this user already sent in parts (`upload-staging.js`), so a restart in the + * middle costs only the part in flight. `write(stream, length)` answers + * `{status, body}`; a staged commit made before answers its stored answer. + */ +function stagedOr(req, write) { + const id = req.query.staged; + if (id === undefined) { + return write(req, req.headers["content-length"]); + } + req.resume(); + return uploadStaging.commit(fsUserKey(req), id, write); +} + +/** + * One part of a resumable upload (`?upload=&offset=&total=`, raw bytes); GET = how + * much arrived (`{received, total, done}`), DELETE = forget it. The bytes stay + * on the host until `/fs/upload` or `/fs/upload-batch` names `?staged=`. + */ +router.get("/:id/fs/upload-part", async (req, res) => { + const t = await gitTargetOr404(req, res); + if (!t) return; + try { + res.json(uploadStaging.status(fsUserKey(req), req.query.upload)); + } catch (err) { + fsFail(res, err); + } +}); +router.post("/:id/fs/upload-part", sameOriginGuard, async (req, res) => { + const t = await gitTargetOr404(req, res); + if (!t) return; + try { + res.json( + await uploadStaging.putPart(fsUserKey(req), req.query.upload, { + offset: req.query.offset, + total: req.query.total, + stream: req, + }) + ); + } catch (err) { + req.resume(); + fsFail(res, err); + } +}); +router.delete("/:id/fs/upload-part", sameOriginGuard, async (req, res) => { + const t = await gitTargetOr404(req, res); + if (!t) return; + try { + res.json(uploadStaging.drop(fsUserKey(req), req.query.upload)); + } catch (err) { + fsFail(res, err); + } +}); + /** * Upload one file, streamed (no body parser buffers it), to `?path=` — any * missing sub-folders are made (a folder upload sends each file with its @@ -2065,17 +2126,17 @@ router.post("/:id/fs/upload", sameOriginGuard, async (req, res) => { const t = await gitTargetOr404(req, res); if (!t) return; try { - const out = await fileExplorer.upload( - t.cwd, - typeof req.query.path === "string" ? req.query.path : "", - req, - { - conflict: req.query.conflict, - length: req.headers["content-length"], - trash: fsTrashFor(req, t), - } - ); - res.status(out.skipped ? 200 : 201).json(out); + const write = async (stream, length) => { + const out = await fileExplorer.upload( + t.cwd, + typeof req.query.path === "string" ? req.query.path : "", + stream, + { conflict: req.query.conflict, length, trash: fsTrashFor(req, t) } + ); + return { status: out.skipped ? 200 : 201, body: out }; + }; + const r = await stagedOr(req, write); + res.status(r.status).json(r.body); } catch (err) { req.resume(); fsFail(res, err); @@ -2091,13 +2152,17 @@ router.post("/:id/fs/upload-batch", sameOriginGuard, async (req, res) => { const t = await gitTargetOr404(req, res); if (!t) return; try { - const out = await fileExplorer.uploadBatch( - t.cwd, - typeof req.query.dir === "string" ? req.query.dir : "", - req, - { conflict: req.query.conflict, trash: fsTrashFor(req, t) } - ); - res.status(201).json(out); + const write = async (stream) => ({ + status: 201, + body: await fileExplorer.uploadBatch( + t.cwd, + typeof req.query.dir === "string" ? req.query.dir : "", + stream, + { conflict: req.query.conflict, trash: fsTrashFor(req, t) } + ), + }); + const r = await stagedOr(req, write); + res.status(r.status).json(r.body); } catch (err) { req.resume(); fsFail(res, err);