diff --git a/src/backend/download-routes.ts b/src/backend/download-routes.ts index 9b2a7de..74670ab 100644 --- a/src/backend/download-routes.ts +++ b/src/backend/download-routes.ts @@ -16,6 +16,56 @@ import { * reading the rest of the file, minutes later for a large one. */ const CANCELLED_CLOSE_GRACE_MS = 2000; +/** Files below this size are read over the session's shared download channel. */ +const DEDICATED_CHANNEL_MIN_BYTES = 8 * 1024 * 1024; + +/** + * One long-lived SFTP channel per session for small downloads. Opening a + * channel per file costs a round trip or more and servers cap them (~10), so + * batch downloads of small files were bound by that. It is separate from + * session.sftp, so a stalled download does not queue browsing behind it on + * the channel (the SSH connection itself is still shared). + */ +const bulkChannels = new WeakMap>(); + +function evictBulkChannel( + session: SSHSession, + channel: Promise | undefined, +) { + // Only the cached channel may be dropped, never a newer replacement. + if (channel && bulkChannels.get(session) === channel) + bulkChannels.delete(session); +} + +function getBulkChannel(session: SSHSession): Promise { + let channel = bulkChannels.get(session); + if (!channel) { + const opened: Promise = openDedicatedSftp(session).then( + (sftp) => { + sftp.once("close", () => evictBulkChannel(session, opened)); + return sftp; + }, + ); + channel = opened; + bulkChannels.set(session, opened); + // A failed open must not be cached. + opened.catch(() => evictBulkChannel(session, opened)); + } + return channel; +} + +/** + * Ends a stalled download channel and uncaches it right away (so the next + * download opens a fresh one) unless a newer channel already replaced it. + */ +function discardBulkChannel( + session: SSHSession, + channel: Promise | undefined, + sftp: SFTPWrapper, +) { + evictBulkChannel(session, channel); + sftp.end(); +} type FileDownloadRoutesDeps = { ctx: PluginContext; @@ -296,20 +346,22 @@ export function registerFileDownloadRoutes( return; } - // A download gets its own channel so that a server stalling on it (e.g. - // on the CLOSE of a cancelled file) cannot block browsing on session.sftp. + // Small files share one download channel (see getBulkChannel). Big files + // get their own so a server stalling on one (e.g. on the CLOSE of a + // cancelled file) cannot hold up the small ones. Normally neither uses + // session.sftp, which stays free for browsing; it is only the fallback + // when no download channel can be opened. + // Only a dedicated channel has something to release; a shared download + // channel outlives the request, so for those this stays a no-op. let release = () => {}; + let bulk: SFTPWrapper | undefined; + let bulkPromise: Promise | undefined; try { let sftp: SFTPWrapper; try { - const dedicated = await openDedicatedSftp(sshConn); - sftp = dedicated; - let released = false; - release = () => { - if (released) return; - released = true; - dedicated.end(); - }; + bulkPromise = getBulkChannel(sshConn); + bulk = await bulkPromise; + sftp = bulk; } catch (err) { ctx.log.warn( `Could not open a download SFTP channel, using the session's: ${sessionId} (${(err as Error).message})`, @@ -324,10 +376,27 @@ export function registerFileDownloadRoutes( ); if (!stats.isFile()) { - release(); return res.status(400).json({ error: "Cannot download directories" }); } + if (stats.size >= DEDICATED_CHANNEL_MIN_BYTES && !res.destroyed) { + try { + const dedicated = await openDedicatedSftp(sshConn); + sftp = dedicated; + bulk = undefined; + let released = false; + release = () => { + if (released) return; + released = true; + dedicated.end(); + }; + } catch (err) { + ctx.log.warn( + `Could not open a download SFTP channel, using the session's: ${sessionId} (${(err as Error).message})`, + ); + } + } + res.setHeader("Content-Length", String(stats.size)); if (res.destroyed) { @@ -338,7 +407,14 @@ export function registerFileDownloadRoutes( readStream.on("close", release); res.on("close", () => { readStream.destroy(); - setTimeout(release, CANCELLED_CLOSE_GRACE_MS).unref(); + setTimeout(() => { + release(); + // `closed` is true once the stream finished, so a normal download + // never gets here. Still open: the server is not answering this + // cancelled file, so retire the shared channel rather than let it jam. + if (bulk && !readStream.closed) + discardBulkChannel(sshConn, bulkPromise, bulk); + }, CANCELLED_CLOSE_GRACE_MS).unref(); }); readStream.on("error", (err) => { if (!res.headersSent) { diff --git a/tests/backend/download-routes-sftp.test.ts b/tests/backend/download-routes-sftp.test.ts index c23b580..d3e9746 100644 --- a/tests/backend/download-routes-sftp.test.ts +++ b/tests/backend/download-routes-sftp.test.ts @@ -117,7 +117,8 @@ it("keeps the session channel responsive while a cancelled download's close is u await once(res, "data"); res.destroy(); await vi.waitFor(() => expect(unansweredCloses).toBe(1)); - expect(sftpChannels).toBe(2); + // browsing channel + shared download channel (stat) + the big file's own + expect(sftpChannels).toBe(3); const answered = new Promise((resolve, reject) => shared.stat("/", (err) => (err ? reject(err) : resolve())), diff --git a/tests/backend/download-routes.test.ts b/tests/backend/download-routes.test.ts index dee7ed6..00da958 100644 --- a/tests/backend/download-routes.test.ts +++ b/tests/backend/download-routes.test.ts @@ -9,22 +9,38 @@ import { type SSHSession, } from "../../src/backend/session.js"; +/** At or above the route's threshold a download gets its own channel. */ +const BIG = 8 * 1024 * 1024; + /** - * `remote` is the channel a download opens for itself; `shared` is the - * session's browsing channel, which a download must leave alone. + * Three channels: `shared` is the session's browsing channel (never used by + * downloads); `bulk` is the per-session download channel small files share + * (the first channel the client opens); `dedicated` is what a big file opens + * for itself (every later channel). */ function setup( - remote: object, owned = true, - { dedicatedFails = false }: { dedicatedFails?: boolean } = {}, + { + size = 5, + dedicatedFails = false, + bulkFails = false, + }: { size?: number; dedicatedFails?: boolean; bulkFails?: boolean } = {}, ) { - const shared = source(); + const shared = source(size); + const bulk = source(size); + const dedicated = source(size); + let opened = 0; const client = { - sftp: vi.fn((cb) => - dedicatedFails + sftp: vi.fn((cb) => { + opened++; + if (opened === 1) + return bulkFails + ? cb(new Error("Channel open failure"), undefined) + : cb(undefined, bulk); + return dedicatedFails ? cb(new Error("Channel open failure"), undefined) - : cb(undefined, remote), - ), + : cb(undefined, dedicated); + }), }; const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; const routes = new Map< @@ -53,168 +69,305 @@ function setup( }, ); const headers = new Map(); - const res = Object.assign(new PassThrough(), { - headersSent: false, - setHeader: (name: string, value: string) => headers.set(name, value), - removeHeader: (name: string) => headers.delete(name), - status: vi.fn().mockReturnThis(), - json: vi.fn(), - }); - const start = () => + const makeRes = () => + Object.assign(new PassThrough(), { + headersSent: false, + setHeader: (name: string, value: string) => headers.set(name, value), + removeHeader: (name: string) => headers.delete(name), + status: vi.fn().mockReturnThis(), + json: vi.fn(), + }); + const res = makeRes(); + const start = (target = res) => routes.get("/downloadFileStream")!( { body: { sessionId: "s", path: "/file" } } as Request, - res as unknown as Response, + target as unknown as Response, ); - return { res, start, headers, shared, client, log }; + return { res, makeRes, start, headers, shared, bulk, dedicated, client, log }; } -function source() { +function source(size = 5) { return Object.assign(new EventEmitter(), { - stat: vi.fn((_path, cb) => cb(undefined, { size: 5, isFile: () => true })), + stat: vi.fn((_path, cb) => cb(undefined, { size, isFile: () => true })), open: vi.fn((_path, _flags, _mode, cb) => cb(undefined, Buffer.from("handle")), ), - read: vi.fn((_handle, buffer, offset, _length, _position, cb) => { - Buffer.from("hello").copy(buffer, offset); - cb(undefined, 5); + read: vi.fn((_handle, buffer, offset, length, position, cb) => { + const bytes = Math.min(length, size - position); + for (let i = 0; i < bytes; i++) + buffer[offset + i] = "hello".charCodeAt((position + i) % 5); + cb(undefined, bytes); }), close: vi.fn((_handle, cb) => cb()), end: vi.fn(), }); } +/** Reads of `channel` stay unanswered until the returned function is called. */ +function holdReads(channel: ReturnType) { + const pending: Array<() => void> = []; + channel.read.mockImplementation((_h, _b, _o, _l, _p, cb) => { + pending.push(() => cb(undefined, 0)); + }); + return () => pending.forEach((reply) => reply()); +} + +async function drain(res: PassThrough) { + let received = 0; + for await (const chunk of res) received += chunk.length; + return received; +} + afterEach(() => { vi.useRealTimers(); }); describe("stream download route", () => { - it("streams the file content with its advertised length", async () => { - const remote = source(); - const { res, start, headers } = setup(remote); + it("streams a small file over the shared download channel, not browsing's", async () => { + const { res, start, headers, shared, bulk, dedicated, client } = setup(); await start(); const chunks: Buffer[] = []; for await (const chunk of res) chunks.push(chunk); expect(Buffer.concat(chunks).toString()).toBe("hello"); expect(headers.get("Content-Length")).toBe("5"); - expect(remote.close).toHaveBeenCalledOnce(); + expect(client.sftp).toHaveBeenCalledOnce(); + expect(bulk.close).toHaveBeenCalledOnce(); + expect(bulk.end).not.toHaveBeenCalled(); + expect(dedicated.stat).not.toHaveBeenCalled(); + expect(shared.stat).not.toHaveBeenCalled(); + expect(shared.open).not.toHaveBeenCalled(); + }); + + it("reuses one download channel for several small files", async () => { + const { res, makeRes, start, client, bulk } = setup(); + const second = makeRes(); + await start(); + await start(second); + expect(await drain(res)).toBe(5); + expect(await drain(second)).toBe(5); + expect(client.sftp).toHaveBeenCalledOnce(); + expect(bulk.open).toHaveBeenCalledTimes(2); + expect(bulk.end).not.toHaveBeenCalled(); + }); + + it("keeps the shared channel after normal downloads, even once the grace timer passes", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const { res, makeRes, start, bulk, client } = setup(); + await start(); + expect(await drain(res)).toBe(5); + // A finished response also emits "close", which arms the grace timer. + vi.advanceTimersByTime(5000); + expect(bulk.end).not.toHaveBeenCalled(); + const next = makeRes(); + await start(next); + expect(await drain(next)).toBe(5); + vi.advanceTimersByTime(5000); + expect(bulk.end).not.toHaveBeenCalled(); + expect(client.sftp).toHaveBeenCalledOnce(); + }); + + it("keeps the shared channel and closes the handle when a small download is cancelled", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const { res, start, bulk, shared, client } = setup(); + const finishRead = holdReads(bulk); + await start(); + await vi.waitFor(() => expect(bulk.read).toHaveBeenCalled()); + res.destroy(); + await once(res, "close"); + finishRead(); + await vi.waitFor(() => expect(bulk.close).toHaveBeenCalledOnce()); + vi.advanceTimersByTime(2000); + expect(bulk.end).not.toHaveBeenCalled(); + expect(shared.end).not.toHaveBeenCalled(); + expect(client.sftp).toHaveBeenCalledOnce(); + }); + + it("retires the shared channel when a cancelled small file's close is never answered", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const { res, makeRes, start, bulk, shared, client } = setup(); + const finishRead = holdReads(bulk); + // Some servers only answer CLOSE after reading the rest of the file. + bulk.close.mockImplementation(() => {}); + await start(); + await vi.waitFor(() => expect(bulk.read).toHaveBeenCalled()); + res.destroy(); + await once(res, "close"); + finishRead(); + await vi.waitFor(() => expect(bulk.close).toHaveBeenCalledOnce()); + expect(bulk.end).not.toHaveBeenCalled(); + vi.advanceTimersByTime(2000); + expect(bulk.end).toHaveBeenCalledOnce(); + expect(shared.end).not.toHaveBeenCalled(); + // The next download gets a fresh channel instead of the jammed one. + const next = makeRes(); + await start(next); + expect(await drain(next)).toBe(5); + expect(client.sftp).toHaveBeenCalledTimes(2); + }); + + it("a late retirement timer does not drop a newer shared channel", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const { res, makeRes, start, bulk, client } = setup(); + const finishRead = holdReads(bulk); + bulk.close.mockImplementation(() => {}); + const second = makeRes(); + await start(); + await start(second); + await vi.waitFor(() => expect(bulk.read).toHaveBeenCalled()); + // First cancellation: its timer retires channel 1. + res.destroy(); + await once(res, "close"); + vi.advanceTimersByTime(1000); + // Second cancellation, 1 s later, so its timer fires after a replacement. + second.destroy(); + await once(second, "close"); + finishRead(); + await vi.waitFor(() => expect(bulk.close).toHaveBeenCalledTimes(2)); + vi.advanceTimersByTime(1000); + expect(bulk.end).toHaveBeenCalledOnce(); + // A new download opens channel 2 and caches it. + const third = makeRes(); + await start(third); + expect(client.sftp).toHaveBeenCalledTimes(2); + // Channel 1's second timer fires now; it must not uncache channel 2. + vi.advanceTimersByTime(2000); + const fourth = makeRes(); + await start(fourth); + expect(client.sftp).toHaveBeenCalledTimes(2); + third.destroy(); + fourth.destroy(); }); - it("does not open a file after the HTTP client disconnected during stat", async () => { - const remote = source(); + it("does not open a file or a dedicated channel after the client left during stat", async () => { + const { res, start, bulk, dedicated, client } = setup(true, { size: BIG }); let finishStat!: () => void; - remote.stat.mockImplementation((_path, cb) => { - finishStat = () => cb(undefined, { size: 5, isFile: () => true }); + bulk.stat.mockImplementation((_path, cb) => { + finishStat = () => cb(undefined, { size: BIG, isFile: () => true }); }); - const { res, start } = setup(remote); const started = start(); - await vi.waitFor(() => expect(remote.stat).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(bulk.stat).toHaveBeenCalledOnce()); res.destroy(); finishStat(); await started; - expect(remote.open).not.toHaveBeenCalled(); - expect(remote.end).toHaveBeenCalledOnce(); + expect(bulk.open).not.toHaveBeenCalled(); + expect(dedicated.open).not.toHaveBeenCalled(); + expect(client.sftp).toHaveBeenCalledOnce(); }); - it("closes the SFTP handle after HTTP disconnect during a read", async () => { - const remote = source(); - let finishRead!: () => void; - remote.read.mockImplementation( - (_h, buffer, offset, length, position, cb) => { - finishRead = () => cb(undefined, 0); - }, - ); - const { res, start } = setup(remote); + it("uses the shared channel just below the threshold and a dedicated one at it", async () => { + const below = setup(true, { size: BIG - 1 }); + await below.start(); + await drain(below.res); + expect(below.client.sftp).toHaveBeenCalledOnce(); + expect(below.dedicated.open).not.toHaveBeenCalled(); + + const at = setup(true, { size: BIG }); + await at.start(); + await drain(at.res); + expect(at.client.sftp).toHaveBeenCalledTimes(2); + expect(at.dedicated.open).toHaveBeenCalledOnce(); + expect(at.bulk.open).not.toHaveBeenCalled(); + }); + + it("closes the dedicated handle after disconnect during a big read", async () => { + const { res, start, dedicated } = setup(true, { size: BIG }); + const finishRead = holdReads(dedicated); await start(); - await new Promise((resolve) => setImmediate(resolve)); + await vi.waitFor(() => expect(dedicated.read).toHaveBeenCalled()); res.destroy(); await once(res, "close"); finishRead(); - await vi.waitFor(() => expect(remote.close).toHaveBeenCalledOnce()); - expect(remote.read).toHaveBeenCalledOnce(); - await vi.waitFor(() => expect(remote.end).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(dedicated.close).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(dedicated.end).toHaveBeenCalledOnce()); }); - it("survives an error event on its own SFTP channel", async () => { - const remote = source(); - const { res, start } = setup(remote); + it("survives an error event on a download channel", async () => { + const { res, start, dedicated, bulk } = setup(true, { size: BIG }); await start(); expect(() => - remote.emit("error", new Error("channel reset")), + dedicated.emit("error", new Error("channel reset")), ).not.toThrow(); + expect(() => bulk.emit("error", new Error("channel reset"))).not.toThrow(); res.destroy(); }); - it("downloads over its own SFTP channel and ends it afterwards", async () => { - const remote = source(); - const { res, start, shared, client } = setup(remote); + it("downloads a big file over its own channel and ends only that channel", async () => { + const { res, start, shared, bulk, dedicated, client } = setup(true, { + size: BIG, + }); await start(); - for await (const _chunk of res); - expect(client.sftp).toHaveBeenCalledOnce(); - expect(remote.stat).toHaveBeenCalledOnce(); + expect(await drain(res)).toBe(BIG); + expect(client.sftp).toHaveBeenCalledTimes(2); + expect(dedicated.open).toHaveBeenCalledOnce(); + expect(bulk.open).not.toHaveBeenCalled(); expect(shared.stat).not.toHaveBeenCalled(); - expect(shared.open).not.toHaveBeenCalled(); - await vi.waitFor(() => expect(remote.end).toHaveBeenCalledOnce()); - expect(shared.end).not.toHaveBeenCalled(); + await vi.waitFor(() => expect(dedicated.end).toHaveBeenCalledOnce()); + expect(bulk.end).not.toHaveBeenCalled(); }); - it("falls back to the session channel when its own cannot be opened", async () => { - const remote = source(); - const { res, start, shared, log } = setup(remote, true, { + it("falls back to the shared download channel when a big file's own cannot be opened", async () => { + const { res, start, bulk, log } = setup(true, { dedicatedFails: true, + size: BIG, }); await start(); - const chunks: Buffer[] = []; - for await (const chunk of res) chunks.push(chunk); - expect(Buffer.concat(chunks).toString()).toBe("hello"); - expect(shared.stat).toHaveBeenCalledOnce(); + expect(await drain(res)).toBe(BIG); + expect(bulk.open).toHaveBeenCalledOnce(); + expect(bulk.end).not.toHaveBeenCalled(); + expect(log.warn).toHaveBeenCalledWith( + expect.stringContaining("Channel open failure"), + ); + }); + + it("falls back to the session channel when no download channel can be opened", async () => { + const { res, start, shared, log } = setup(true, { bulkFails: true }); + await start(); + expect(await drain(res)).toBe(5); + expect(shared.open).toHaveBeenCalledOnce(); expect(shared.end).not.toHaveBeenCalled(); expect(log.warn).toHaveBeenCalledWith( expect.stringContaining("Channel open failure"), ); }); - it("ends its channel when the server does not answer the cancelled file's close", async () => { + it("ends a big file's channel when the server does not answer the cancelled file's close", async () => { vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); - const remote = source(); - let finishRead!: () => void; - remote.read.mockImplementation((_h, _b, _o, _l, _p, cb) => { - finishRead = () => cb(undefined, 0); - }); - // Some servers only answer CLOSE after reading the rest of the file. - remote.close.mockImplementation(() => {}); - const { res, start } = setup(remote); + const { res, start, bulk, dedicated } = setup(true, { size: BIG }); + const finishRead = holdReads(dedicated); + dedicated.close.mockImplementation(() => {}); await start(); - await new Promise((resolve) => setImmediate(resolve)); + await vi.waitFor(() => expect(dedicated.read).toHaveBeenCalled()); res.destroy(); await once(res, "close"); finishRead(); - await vi.waitFor(() => expect(remote.close).toHaveBeenCalledOnce()); - expect(remote.end).not.toHaveBeenCalled(); + await vi.waitFor(() => expect(dedicated.close).toHaveBeenCalledOnce()); + expect(dedicated.end).not.toHaveBeenCalled(); vi.advanceTimersByTime(2000); - expect(remote.end).toHaveBeenCalledOnce(); + expect(dedicated.end).toHaveBeenCalledOnce(); + expect(bulk.end).not.toHaveBeenCalled(); }); - it("removes the file length before returning an early stream error as JSON", async () => { - const remote = source(); - remote.open.mockImplementation((_p, _f, _m, cb) => + it("removes the file length before returning an early stream error as JSON, keeping the channel", async () => { + const { res, start, headers, bulk } = setup(); + bulk.open.mockImplementation((_p, _f, _m, cb) => cb(new Error("permission denied"), undefined), ); - const { res, start, headers } = setup(remote); await start(); await vi.waitFor(() => expect(res.status).toHaveBeenCalledWith(500)); expect(headers.has("Content-Length")).toBe(false); expect(res.json).toHaveBeenCalledWith({ error: "Download failed: permission denied", }); + expect(bulk.end).not.toHaveBeenCalled(); res.destroy(); }); it("refuses another user's SSH session before reading a file", async () => { - const remote = source(); - const { res, start } = setup(remote, false); + const { res, start, bulk, client } = setup(false); await start(); expect(res.status).toHaveBeenCalledWith(403); - expect(remote.stat).not.toHaveBeenCalled(); + expect(bulk.stat).not.toHaveBeenCalled(); + expect(client.sftp).not.toHaveBeenCalled(); res.destroy(); }); });