diff --git a/docs/configuration.md b/docs/configuration.md index f87f148e..fb0448bc 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -231,15 +231,17 @@ Must be greater than 0. Defaults to 15. -### controller.transfer_idle_timeout +### controller.transfer_min_progress -Time in seconds an upload to or download from the controller can go without any data being transferred before the controller closes it. -This applies to saves, mods and exports uploaded to the controller and to proxy streams through it. -There is no limit on how long a transfer can take as long as it keeps making progress. -Time spent waiting for a proxy stream to start also counts, so this should be higher than controller.proxy_stream_timeout. -Must be greater than 0. +Amount in KiB a request to the controller has to move every 30 seconds, requests moving less are closed. +This closes uploads and downloads that have stalled or are sent too slowly to ever finish, while there is no limit on how long a transfer making progress can take. +It applies to saves, mods and exports uploaded to the controller, proxy streams through it, and the body of any other request. +Time spent waiting for a proxy stream to start is not counted, see controller.proxy_stream_timeout. +Data is passed on in chunks of up to 64 KiB and progress is counted per chunk, so a proxy stream moving less than about that in 30 seconds may also be closed. +Set to 0 for no limit. +Must be 0 or greater. -Defaults to 60. +Defaults to 30. ### controller.factorio_username diff --git a/packages/controller/src/routes.ts b/packages/controller/src/routes.ts index c61fe9b0..9c5121d5 100644 --- a/packages/controller/src/routes.ts +++ b/packages/controller/src/routes.ts @@ -417,6 +417,9 @@ export interface ProxyStream { filename: string | null; events: events.EventEmitter; timeout: ReturnType; + closed: boolean; + /** Bytes the receiver has taken, set once the stream is flowing */ + progress?: () => number; } export async function createProxyStream(app: Application): Promise { @@ -433,8 +436,10 @@ export async function createProxyStream(app: Application): Promise timeout: setTimeout(() => { stream.events.emit("timeout"); }, app.locals.controller.config.get("controller.proxy_stream_timeout") * 1000), + closed: false, }; stream.events.on("close", () => { + stream.closed = true; clearTimeout(stream.timeout); app.locals.streams.delete(id); }); @@ -450,15 +455,17 @@ export async function createProxyStream(app: Application): Promise * * Node closes requests that have not been received in full after 5 minutes * by default, which would cut off large uploads over slow connections. - * Transfers are instead closed when they stop making progress. + * Requests are instead closed by guardRequest when they move too little. + * Turning the limit off also turns off the limit on receiving the headers, + * which guardRequest can't cover, so that is set back to its default. */ -export const httpServerOptions = { requestTimeout: 0 }; +export const httpServerOptions = { requestTimeout: 0, headersTimeout: 60_000 }; /** Time an upload is read and thrown away after it was answered, before its connection is closed */ const answeredUploadDrainMs = 1000; /** - * Guard a route transferring files against wasted and stalled transfers + * Guard requests against wasted and stalled transfers * * Node reads and discards the body of a request answered before it was * read to keep the connection open. A client, or a proxy in front of the @@ -466,35 +473,65 @@ const answeredUploadDrainMs = 1000; * so the connection is closed if the body hasn't ended shortly after the * response. Closing it right away could make the client lose the response. * - * The connection is also closed if no data was sent or received on it for - * controller.transfer_idle_timeout, so that a transfer which stopped - * making progress does not linger while a slow one can take its time. - * @param req - Request to guard. - * @param res - Response to the request. - * @param next - Next handler for the route. + * The connection is also closed if the request moved less than + * controller.transfer_min_progress during a check interval, so that a + * request which stalled or is trickled in does not stay open forever while + * a slow one can take the time it needs. For a request body this is what + * was received of it. For both ends of a proxy stream it is what the + * receiver has taken, as the sender is held back when the receiver is the + * slow one. Time spent waiting for a proxy stream to start is not counted, + * controller.proxy_stream_timeout limits that. + * @param checkIntervalMs - Interval to check the progress of requests on. + * @returns Express middleware. */ -function guardTransfer(req: Request, res: Response, next: any) { - const socket = req.socket; - const idleTimeout = req.app.locals.controller.config.get("controller.transfer_idle_timeout"); - // Node sets the timeout for the next request on the connection itself. - res.setTimeout(idleTimeout * 1000, () => { - logger.warn(`Closing ${req.method} ${req.path} after ${idleTimeout} seconds without progress`); - socket.destroy(); - }); - res.once("finish", () => { - if (req.complete || socket.destroyed) { - return; - } - const timeout = setTimeout(() => socket.destroy(), answeredUploadDrainMs); - const onClose = () => clearTimeout(timeout); - socket.once("close", onClose); - // The connection may be kept open for more requests, so don't leave a listener on it - req.once("end", () => { - clearTimeout(timeout); - socket.off("close", onClose); +function guardRequest(checkIntervalMs: number) { + return function(req: Request, res: Response, next: any) { + const socket = req.socket; + res.once("finish", () => { + if (req.complete || socket.destroyed) { + return; + } + const timeout = setTimeout(() => socket.destroy(), answeredUploadDrainMs); + const onClose = () => clearTimeout(timeout); + socket.once("close", onClose); + // The connection may be kept open for more requests, so don't leave a listener on it + req.once("end", () => { + clearTimeout(timeout); + socket.off("close", onClose); + }); }); - }); - next(); + + const received = () => socket.bytesRead; + let measure: (() => number) | undefined = received; + let last = received(); + const check = setInterval(() => { + const stream: ProxyStream | undefined = res.locals.proxyStream; + let current: (() => number) | undefined; + if (stream && !stream.closed) { + current = stream.progress; + } else if (!req.complete) { + current = received; + } + const minProgress = req.app.locals.controller.config.get("controller.transfer_min_progress"); + if (current !== measure || !current || !minProgress) { + // What to measure changed, start over from here. + measure = current; + last = current ? current() : 0; + return; + } + const progress = current(); + if (progress - last < minProgress * 1024) { + logger.warn( + `Closing ${req.method} ${req.path}, it moved less than ${minProgress} KiB ` + + `in ${checkIntervalMs / 1000} seconds` + ); + socket.destroy(); + } + last = progress; + }, checkIntervalMs); + res.once("close", () => clearInterval(check)); + next(); + }; } async function putStream(req: Request<{ id: string }>, res: Response) { @@ -503,6 +540,7 @@ async function putStream(req: Request<{ id: string }>, res: Response) { res.sendStatus(404); return; } + res.locals.proxyStream = stream; stream.source = req; stream.mime = req.get("Content-Type"); stream.size = req.get("Content-Length"); @@ -525,8 +563,11 @@ function startStream(res: Response, stream: ProxyStream) { res.append("Content-Disposition", "attachment"); } const source = stream.source!; + const socket = res.socket!; source.pipe(res); stream.flowing = true; + // Counts completed writes, so data waiting in a buffer is not progress. + stream.progress = () => socket.bytesWritten - socket.writableLength; // Pipe does not pass on a source that was cut short, the receiver would wait forever. const onSourceClose = () => { if (!res.writableEnded) { @@ -548,6 +589,7 @@ async function getStream(req: Request<{ id: string }>, res: Response) { res.sendStatus(404); return; } + res.locals.proxyStream = stream; if (stream.source) { startStream(res, stream); @@ -724,6 +766,7 @@ async function uploadSave(req: Request, res: Response) { } let proxyStream = await createProxyStream(req.app); + res.locals.proxyStream = proxyStream; proxyStream.source = file.stream; proxyStream.mime = file.mime; let timeout = new Promise((_, reject) => { @@ -813,7 +856,8 @@ async function uploadMod(req: Request, res: Response) { } -export function addRouteHandlers(app: Application) { +export function addRouteHandlers(app: Application, progressCheckIntervalMs = 30_000) { + app.use(guardRequest(progressCheckIntervalMs)); app.get("/metrics", validateUserToken, (req:Request, res:Response, next:any) => getMetrics(req, res).catch(next) @@ -833,25 +877,20 @@ export function addRouteHandlers(app: Application) { res.json({ valid: true }); }); app.put("/api/upload-export", - guardTransfer, validateHostToken, (req:Request, res:Response, next:any) => uploadExport(req, res).catch(next) ); app.put("/api/stream/:id", - guardTransfer, (req:Request<{ id: string }>, res:Response, next:any) => putStream(req, res).catch(next) ); app.get("/api/stream/:id", - guardTransfer, (req:Request<{ id: string }>, res:Response, next:any) => getStream(req, res).catch(next) ); app.post("/api/upload-save", - guardTransfer, validateUserToken, (req:Request, res:Response, next:any) => uploadSave(req, res).catch(next) ); app.post("/api/upload-mod", - guardTransfer, validateUserToken, (req:Request, res:Response, next:any) => uploadMod(req, res).catch(next) ); diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 06a865ce..e71007b6 100644 --- a/packages/lib/src/config/definitions.ts +++ b/packages/lib/src/config/definitions.ts @@ -31,7 +31,7 @@ export interface ControllerConfigFields { "controller.max_subscription_filters": number; "controller.system_metrics_interval": number; "controller.proxy_stream_timeout": number; - "controller.transfer_idle_timeout": number; + "controller.transfer_min_progress": number; "controller.factorio_username": string | null; "controller.factorio_token": string | null; "controller.share_factorio_credentials_with_hosts": boolean; @@ -216,12 +216,12 @@ export class ControllerConfig extends classes.Config { initialValue: 15, validator: validators.greaterThanZero, }, - "controller.transfer_idle_timeout": { - title: "Transfer Idle Timeout", - description: "Time in seconds an upload or download can go without progress before it's closed.", + "controller.transfer_min_progress": { + title: "Transfer Minimum Progress", + description: "KiB an upload or download has to move every 30 seconds to not be closed, 0 for no limit.", type: "number", - initialValue: 60, - validator: validators.greaterThanZero, + initialValue: 30, + validator: validators.greaterThanEqualZero, }, "controller.session_timeout": { title: "Session Timeout", diff --git a/test/controller/routes.js b/test/controller/routes.js index dd764e68..32669064 100644 --- a/test/controller/routes.js +++ b/test/controller/routes.js @@ -22,7 +22,8 @@ describe("controller/src/routes", function() { let port; beforeEach(async function() { controller = new mock.MockController(); - routes.addRouteHandlers(controller.app); + // Check the progress of requests every 500 ms instead of every 30 seconds. + routes.addRouteHandlers(controller.app, 500); server = http.createServer(routes.httpServerOptions, controller.app); server.listen(0, "localhost"); await events.once(server, "listening"); @@ -149,8 +150,36 @@ describe("controller/src/routes", function() { request.destroy(); await assert.rejects(response.arrayBuffer()); }); + /** + * PUT to a stream at a steady pace until the connection closes or the body is sent + * @param {string} id - Stream to send to. + * @param {number} chunkSize - Bytes to send every 100 ms. + * @param {number} chunks - Number of chunks the body consists of. + * @returns {Promise} status responded with, if any. + */ + async function putPaced(id, chunkSize, chunks) { + const request = http.request( + `${endpoint}/${id}`, { method: "PUT", headers: { "Content-Length": chunkSize * chunks } } + ); + let status; + request.on("response", response => { + status = response.statusCode; + response.resume(); + }); + request.on("error", () => {}); + const closed = new Promise(resolve => request.on("close", resolve)); + for (let i = 0; i < chunks && !request.closed; i++) { + request.write(Buffer.alloc(chunkSize)); + await lib.wait(100); + } + if (!request.closed) { + request.end(); + } + await closed; + return status; + } it("should close a transfer that stops making progress", async function() { - controller.mockConfigEntries.set("controller.transfer_idle_timeout", 0.5); + controller.mockConfigEntries.set("controller.transfer_min_progress", 1); const stream = await routes.createProxyStream(controller.app); const request = http.request( `${endpoint}/${stream.id}`, { method: "PUT", headers: { "Content-Length": 1024 * 1024 } } @@ -165,24 +194,58 @@ describe("controller/src/routes", function() { await closed; assert(Date.now() - start < 1500, `closed after ${Date.now() - start} ms`); }); + it("should close a transfer that is trickled in", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 4); + const stream = await routes.createProxyStream(controller.app); + const start = Date.now(); + const upload = putPaced(stream.id, 200, 50); + const response = await fetch(`${endpoint}/${stream.id}`); + assert.equal(response.status, 200); + await assert.rejects(response.arrayBuffer()); + assert.equal(await upload, undefined); + assert(Date.now() - start < 2000, `closed after ${Date.now() - start} ms`); + }); it("should not close a slow transfer that makes progress", async function() { this.timeout(10000); - controller.mockConfigEntries.set("controller.transfer_idle_timeout", 0.5); + controller.mockConfigEntries.set("controller.transfer_min_progress", 1); const stream = await routes.createProxyStream(controller.app); - const request = http.request( - `${endpoint}/${stream.id}`, { method: "PUT", headers: { "Content-Length": 10 * 1024 } } - ); - const uploaded = events.once(request, "response"); + const upload = putPaced(stream.id, 1024, 20); const download = fetch(`${endpoint}/${stream.id}`).then(response => response.arrayBuffer()); - for (let i = 0; i < 10; i++) { - request.write(Buffer.alloc(1024)); - await lib.wait(200); - } - request.end(); - const [response] = await uploaded; - response.resume(); - assert.equal(response.statusCode, 200); - assert.equal((await download).byteLength, 10 * 1024); + assert.equal(await upload, 200); + assert.equal((await download).byteLength, 20 * 1024); + }); + it("should not close a transfer held back by a slow receiver", async function() { + this.timeout(20000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 16); + const size = 4 * 1024 * 1024; + const stream = await routes.createProxyStream(controller.app); + const upload = fetch(`${endpoint}/${stream.id}`, { method: "PUT", body: Buffer.alloc(size) }); + const received = await new Promise((resolve, reject) => { + http.get(`${endpoint}/${stream.id}`, response => { + let bytes = 0; + response.on("data", chunk => { + bytes += chunk.length; + response.pause(); + setTimeout(() => response.resume(), 50); + }); + response.on("end", () => resolve(bytes)); + response.on("error", reject); + }).on("error", reject); + }); + assert.equal(received, size); + assert.equal((await upload).status, 200); + }); + it("should not count the time waiting for the stream to start", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.proxy_stream_timeout", 3); + controller.mockConfigEntries.set("controller.transfer_min_progress", 1); + const stream = await routes.createProxyStream(controller.app); + const download = fetch(`${endpoint}/${stream.id}`); + await lib.wait(1500); + const upload = await fetch(`${endpoint}/${stream.id}`, { method: "PUT", body: "test content" }); + assert.equal(upload.status, 200); + assert.equal(await (await download).text(), "test content"); }); it("should passthrough a stream", async function() { let stream; @@ -1169,6 +1232,65 @@ describe("controller/src/routes", function() { } }); }); + describe("guardRequest()", function() { + async function trickle(url, headers) { + const request = http.request(url, { method: "POST", headers: { ...headers, "Content-Length": 2 ** 20 } }); + let status; + request.on("response", response => { + status = response.statusCode; + response.resume(); + }); + request.on("error", () => {}); + const closed = new Promise(resolve => request.on("close", resolve)); + const start = Date.now(); + while (!request.closed && Date.now() - start < 5000) { + request.write(Buffer.alloc(200)); + await lib.wait(100); + } + request.destroy(); + await closed; + return { status, duration: Date.now() - start }; + } + it("should keep the limit on receiving headers", function() { + assert.equal(server.requestTimeout, 0); + assert.equal(server.headersTimeout, 60000); + }); + it("should close an upload that is trickled in", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 4); + const modsDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "clusterio-mods-")); + try { + controller.mockConfigEntries.set("controller.mods_directory", modsDirectory); + const result = await trickle(`http://localhost:${port}/api/upload-mod?filename=mod_1.0.0.zip`, { + "Content-Type": "application/zip", + "X-Access-Token": controller.users.signUserToken({ id: "test" }), + }); + assert.equal(result.status, undefined); + assert(result.duration < 2000, `closed after ${result.duration} ms`); + for (let i = 0; i < 50 && (await fs.readdir(modsDirectory)).length; i++) { + await lib.wait(20); + } + assert.deepEqual(await fs.readdir(modsDirectory), []); + } finally { + await fs.rm(modsDirectory, { recursive: true, force: true }); + } + }); + it("should close a request to another route that is trickled in", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 4); + controller.app.post("/test", express.raw({ type: "*/*", limit: "2mb" }), (req, res) => res.sendStatus(200)); + const result = await trickle(`http://localhost:${port}/test`, { "Content-Type": "text/plain" }); + assert.equal(result.status, undefined); + assert(result.duration < 2000, `closed after ${result.duration} ms`); + }); + it("should not close a trickled in request when the limit is 0", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 0); + controller.app.post("/test", express.raw({ type: "*/*", limit: "2mb" }), (req, res) => res.sendStatus(200)); + const result = await trickle(`http://localhost:${port}/test`, { "Content-Type": "text/plain" }); + assert(result.duration >= 5000, `closed after ${result.duration} ms`); + }); + }); describe("/api/upload-export", function() { let endpoint; beforeEach(function() { diff --git a/test/mock.js b/test/mock.js index b64e0b0b..32b5270b 100644 --- a/test/mock.js +++ b/test/mock.js @@ -175,7 +175,7 @@ export class MockController { ["controller.public_url", "test"], ["controller.auth_secret", "TestSecretDoNotUse"], ["controller.proxy_stream_timeout", 1], - ["controller.transfer_idle_timeout", 10], + ["controller.transfer_min_progress", 30], ["controller.metrics_timeout", 1], ["controller.host_pairing_timeout", 3600], ["controller.host_pairing_max_pending", 50],