feat(files): uploads survive a dashboard restart
An upload through the File Explorer is now sent in 8 MB parts that the
server stages on the host (<data dir>/upload-staging/<user>/<id>/,
server/lib/upload-staging.js) and writes only once every byte is there,
through the same upload/uploadBatch code and guards (?staged=<id>). A
restart, an update or a crash pauses it: the browser waits up to 3 min,
asks how much arrived (GET /fs/upload-part) and sends the rest; the dock
says the dashboard is restarting meanwhile. A commit's answer is stored
and replayed, never written twice; a refused one (EEXISTS) keeps the
bytes for a retry with another conflict policy.
A file uploaded into a member's folder is now written to a hidden temp
name and moved onto its name only when whole: write-as cannot tell a
dropped connection from the end of a file, and `replace` left a truncated
file under the real name when the server died mid-upload.
New: GET|POST|DELETE /api/{lanes/:id,homes/:who}/fs/upload-part; 409
EOFFSET/EINCOMPLETE/EBUSY, 429 ETOOMANY. Verified on a sandbox server:
SIGTERM mid-upload of a 60 MB file in a real browser, resumed and
sha256-identical.
This commit is contained in:
@@ -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<slot>/`, 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-<id>` is slot 0 and the only one `run_id`/`status` ever record; `agenthub-lane-<id>-<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: `<data dir>/trash/<user id>/`), 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 `<data dir>/trash/<user id>/`, 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: `<data dir>/trash/<user id>/`), 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 `<data dir>/trash/<user id>/`, items tagged `store`). **An upload survives a dashboard restart because its bytes are staged first** (`server/lib/upload-staging.js`): parts go to `<data dir>/upload-staging/<user>/<id>/` on the HOST, keyed by the caller's user and a browser-made id (never a path), and `?staged=<id>` 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-<id>` 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 `<data dir>/artifacts/<random>/` 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.
|
||||
|
||||
@@ -287,6 +287,11 @@ function UploadChip({ job }: { job: UploadJob }) {
|
||||
{isArmed("stop") && !stopping && (
|
||||
<p className="text-[11px] text-status-danger">{t("uploads.stopConfirm")}</p>
|
||||
)}
|
||||
{job.waiting && !stopping && (
|
||||
<p className="text-[11px] text-status-warning" role="status" data-testid="upload-waiting">
|
||||
{t("uploads.waiting")}
|
||||
</p>
|
||||
)}
|
||||
<p className="truncate text-[11px] text-fg-muted" title={job.folder}>
|
||||
{job.folder}
|
||||
{job.name ? ` · ${job.name}` : ""}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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.",
|
||||
|
||||
@@ -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.",
|
||||
|
||||
@@ -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=<id>`; 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ĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
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;
|
||||
});
|
||||
});
|
||||
+179
-17
@@ -1188,7 +1188,9 @@ const sendRaw = <T>(
|
||||
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 = <T>(
|
||||
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<void>((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<T>(
|
||||
place: FsPlace,
|
||||
commitUrl: (id: string) => string,
|
||||
payload: Blob,
|
||||
{ upload, onProgress, onWait, signal }: StagedSendOpts
|
||||
): Promise<T> {
|
||||
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<string, string>) =>
|
||||
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<T>(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"), {
|
||||
|
||||
@@ -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ĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
@@ -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<Pick<UploadJob, "name" | "done" | "sent">>) => set(id, patch),
|
||||
update: (patch: Partial<Pick<UploadJob, "name" | "done" | "sent" | "waiting">>) =>
|
||||
set(id, patch),
|
||||
end: () => {
|
||||
controllers.delete(id);
|
||||
uploads = uploads.filter((u) => u.id !== id);
|
||||
|
||||
+33
@@ -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=<path>&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
|
||||
`<data dir>/upload-staging/<user>/<id>/` 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=<id>` or
|
||||
`upload-batch?…&staged=<id>` 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
|
||||
`.<name>.agenthub-upload-<hex>` 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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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ĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
@@ -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),
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
* `<data dir>/upload-staging/<user key>/<id>/` (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`).
|
||||
*
|
||||
* `<id>` 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ĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
"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 (`<key>/<id>`) 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,
|
||||
};
|
||||
@@ -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 (`<data dir>/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." } },
|
||||
|
||||
+63
-18
@@ -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=<id>` 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;
|
||||
|
||||
+83
-18
@@ -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=<id>` — 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=<id>`.
|
||||
*/
|
||||
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);
|
||||
|
||||
Reference in New Issue
Block a user