Close requests moving too little instead of idle ones

Replace controller.transfer_idle_timeout with
controller.transfer_min_progress, the amount a request has to move
every 30 seconds, so a request trickled in is closed too. The check
now runs for every route.

Both ends of a proxy stream are judged on what the receiver has taken,
so a sender held back by a slow receiver is not closed.

Keep the limit on receiving headers, which turning off the request
timeout also turned off.

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

packages/controller/src/routes.ts

@@ -417,6 +417,9 @@ export interface ProxyStream {
 	filename: string | null;
 	events: events.EventEmitter<ProxyStreamEvents>;
 	timeout: ReturnType<typeof setTimeout>;
+	closed: boolean;
+	/** Bytes the receiver has taken, set once the stream is flowing */
+	progress?: () => number;
 }
 
 export async function createProxyStream(app: Application): Promise<ProxyStream> {
@@ -433,8 +436,10 @@ export async function createProxyStream(app: Application): Promise<ProxyStream>
 		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<ProxyStream>
  *
  * 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<never>((_, 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)
 	);

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<ControllerConfigFields> {
 			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",

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<number|undefined>} 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() {

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],