Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions locales/en.json
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,8 @@
"newFolder": "New Folder",
"rename": "Rename",
"uploading": "Uploading...",
"externalEditorTextOnly": "Only text files can be opened externally, because saves are written back as text.",
"uploadFilesDone": "{{done}} / {{total}} files",
"uploadingFile": "Uploading {{name}}...",
"uploadingChunk": "Uploading {{name}}: {{percent}}% (part {{chunk}} of {{total}})",
"fileName": "File Name",
Expand Down
82 changes: 27 additions & 55 deletions src/backend/content-routes.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { pipelineWriteToSftp } from "./sftp-pipelined-write.js";
import { randomUUID } from "node:crypto";
import type { Express, Request, Response } from "express";
import Busboy from "busboy";
Expand Down Expand Up @@ -1372,9 +1373,9 @@ export function registerFileContentRoutes(
fileStream.resume();
return;
}
const writeStream = sftp.createWriteStream(fullPath);
const upload = pipelineWriteToSftp(sftp, fullPath, fileStream);
const removePartialFile = () => {
writeStream.destroy();
upload.abort();
sftp.unlink(fullPath, (err) => {
if (err) {
ctx.log.error(
Expand All @@ -1386,60 +1387,31 @@ export function registerFileContentRoutes(
};
destroyUpload = removePartialFile;

writeStream.on("error", (err) => {
ctx.log.error("SFTP write stream error during upload:", err);
if (!resolved && !requestAborted) {
upload.done.then(
() => {
if (resolved || requestAborted) return;
resolved = true;
res
.status(500)
.json({ error: `Upload failed: ${err.message}` });
}
});

const completeUpload = () => {
if (resolved || requestAborted) return;
resolved = true;
ctx.log.info(
`Streaming file upload completed: ${sessionId} (${fullPath}, duration=${Date.now() - uploadStartTime}ms)`,
);
res.json({
message: "File uploaded successfully",
path: fullPath,
toast: {
type: "success",
message: `File uploaded: ${fullPath}`,
},
});
};

let sourceEnded = false;
fileStream.on("end", () => {
sourceEnded = true;
});
writeStream.on("finish", completeUpload);
// ssh2's SFTP WriteStream destroys itself inside _final (autoClose),
// and a Writable destroyed before its final callback never emits
// 'finish' on current Node. 'close' fires once the handle has been
// closed after the last write was acknowledged, so it is the
// completion signal that reliably arrives -- but only trust it
// when the whole body was consumed and nothing aborted the upload.
writeStream.on("close", () => {
if (sourceEnded) completeUpload();
});

fileStream.on("error", (err) => {
ctx.log.error("File read stream error during upload:", err);
writeStream.destroy();
if (!resolved && !requestAborted) {
resolved = true;
res
.status(500)
.json({ error: `Upload stream error: ${err.message}` });
}
});

(fileStream as NodeJS.ReadableStream).pipe(
writeStream as unknown as NodeJS.WritableStream,
ctx.log.info(
`Streaming file upload completed: ${sessionId} (${fullPath}, duration=${Date.now() - uploadStartTime}ms)`,
);
res.json({
message: "File uploaded successfully",
path: fullPath,
toast: {
type: "success",
message: `File uploaded: ${fullPath}`,
},
});
},
(err: Error) => {
ctx.log.error("SFTP write error during upload:", err);
if (!resolved && !requestAborted) {
resolved = true;
res
.status(500)
.json({ error: `Upload failed: ${err.message}` });
}
},
);
})
.catch((err: Error) => {
Expand Down
98 changes: 98 additions & 0 deletions src/backend/sftp-pipelined-write.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
type SFTPWrapper = import("ssh2").SFTPWrapper;

const WRITE_CHUNK = 64 * 1024;
const MAX_IN_FLIGHT = 64;

/**
* Streams `source` into a remote file with many WRITE requests in flight.
* ssh2's createWriteStream waits for each ack before the next write, which
* caps throughput at one chunk per round trip.
*/
export function pipelineWriteToSftp(
sftp: SFTPWrapper,
path: string,
source: NodeJS.ReadableStream,
): { done: Promise<number>; abort: () => void } {
let abortHook: () => void = () => {};

const done = new Promise<number>((resolve, reject) => {
let handle: Buffer | undefined;
let inFlight = 0;
let ended = false;
let settled = false;
let written = 0;
let position = 0;
const queue: Buffer[] = [];

const closeHandle = (cb: (err?: Error) => void) => {
if (!handle) return cb();
const h = handle;
handle = undefined;
sftp.close(h, (err) => cb(err ?? undefined));
};
const fail = (err: Error) => {
if (settled) return;
settled = true;
queue.length = 0;
source.removeListener("data", onData);
// Drain the rest of the body so the HTTP request can finish cleanly.
source.resume();
closeHandle(() => reject(err));
};
const maybeFinish = () => {
if (settled || !ended || inFlight > 0 || queue.length > 0) return;
settled = true;
closeHandle((err) => (err ? reject(err) : resolve(written)));
};
// Dispatch queued bytes while fewer than MAX_IN_FLIGHT writes are pending.
const pump = () => {
while (
!settled &&
handle &&
queue.length > 0 &&
inFlight < MAX_IN_FLIGHT
) {
const buf = queue[0];
const len = Math.min(WRITE_CHUNK, buf.length);
const piece = buf.subarray(0, len);
if (len === buf.length) queue.shift();
else queue[0] = buf.subarray(len);
inFlight++;
sftp.write(handle, piece, 0, len, position, (err) => {
inFlight--;
if (err) return fail(err);
written += len;
pump();
maybeFinish();
});
position += len;
}
if (!settled && !ended && queue.length === 0) source.resume();
};
function onData(chunk: Buffer) {
source.pause();
queue.push(chunk);
pump();
}

abortHook = () => fail(new Error("Upload aborted"));
// Attached before open() so an early source error rejects instead of throwing.
source.on("error", fail);
source.once("end", () => {
ended = true;
maybeFinish();
});

sftp.open(path, "w", (err, h) => {
if (err) return fail(err);
if (settled) {
sftp.close(h, () => {});
return;
}
handle = h;
source.on("data", onData);
});
});

return { done, abort: () => abortHook() };
}
Loading