Stop reading rejected save uploads to the controller

When a host pushes a save to a stream that is gone, or that timed out
before anyone downloaded it, the controller answers 404 or 500 without
reading the body. Node then reads and throws away the rest of the body
to keep the connection open, so a proxy in front of the controller that
keeps passing the upload on made the whole save get uploaded for
nothing. Close the connection if the body hasn't ended a second after
the response.

Document turning off request buffering in reverse proxies, as a
buffering proxy holds the upload back until all of it has arrived.

docs/setting-up-tls.md

@@ -19,6 +19,12 @@ Additianlly WebSocket connections to `/api/socket` also needs to be forwarded.
 Typically this requires explicit configuration and support for forwarding WebSocket connections.
 If the interface hosted under a sub-path the `/api/socket` address is relative to the sub-path and the sub-path also needs to be stripped out when forwarded.
 
+Saves are sent from hosts to the web interface through the controller as they are uploaded to `/api/stream/`.
+If the proxy buffers request bodies it holds back a host's upload until all of it has arrived.
+A save that takes longer than `controller.proxy_stream_timeout` to upload then fails, after the whole save was sent.
+Turn request buffering off for the proxy, with Nginx this is `proxy_request_buffering off;`.
+Nginx also limits request bodies to 1 MB by default, set `client_max_body_size` high enough for the saves and mods you upload.
+
 You can also optionally set the controller to listen only on the loopback interface to prevent it being directly connected to from the outside.
 
     npx clusteriocontroller config set controller.bind_address 127.0.0.1

packages/controller/src/routes.ts

@@ -344,10 +344,36 @@ export async function createProxyStream(app: Application): Promise<ProxyStream>
 	return stream;
 }
 
+/** Time a rejected upload is read and thrown away before its connection is closed */
+const rejectedUploadDrainMs = 1000;
+
+/**
+ * Reject an upload without reading the rest of its body
+ *
+ * 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
+ * controller, that keeps sending would upload the whole body for nothing,
+ * 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.
+ * @param req - The upload.
+ * @param res - Response to the upload.
+ * @param status - HTTP status to respond with.
+ */
+function rejectUpload(req: Request, res: Response, status: number) {
+	res.sendStatus(status);
+	if (req.complete) {
+		return;
+	}
+	const socket = req.socket;
+	const timeout = setTimeout(() => socket.destroy(), rejectedUploadDrainMs);
+	req.once("end", () => clearTimeout(timeout));
+	socket.once("close", () => clearTimeout(timeout));
+}
+
 async function putStream(req: Request<{ id: string }>, res: Response) {
 	const stream = req.app.locals.streams.get(req.params.id);
 	if (!stream || stream.source) {
-		res.sendStatus(404);
+		rejectUpload(req, res, 404);
 		return;
 	}
 	stream.source = req;
@@ -357,8 +383,7 @@ async function putStream(req: Request<{ id: string }>, res: Response) {
 	stream.events.emit("source");
 	stream.events.on("close", () => {
 		if (!stream.flowing) {
-			req.resume();
-			res.sendStatus(500);
+			rejectUpload(req, res, 500);
 		} else {
 			res.sendStatus(200);
 		}

test/controller/routes.js

@@ -42,6 +42,58 @@ describe("controller/src/routes", function() {
 			let response = await fetch(`${endpoint}/${stream.id}`);
 			assert.equal(response.status, 500);
 		});
+		/**
+		 * PUT a body at about 6 MiB/s whether or not the server reads it, the way
+		 * a proxy in front of the controller passes on an upload, until the
+		 * connection closes, and count what the server read.
+		 * @param {string} url - URL to PUT to.
+		 * @param {number} size - Size of the body in bytes.
+		 * @returns {Promise<{ status: number|undefined, serverBytesRead: number }>} result
+		 */
+		async function putLargeBody(url, size) {
+			const sockets = [];
+			const onConnection = socket => sockets.push(socket);
+			server.on("connection", onConnection);
+			const request = http.request(url, { method: "PUT", headers: { "Content-Length": size } });
+			let status;
+			request.on("response", response => {
+				status = response.statusCode;
+				response.on("error", () => {});
+				response.resume();
+			});
+			// The server may close the connection while the body is being sent
+			request.on("error", () => {});
+			const closed = new Promise(resolve => request.on("close", resolve));
+			const chunk = Buffer.alloc(64 * 1024);
+			for (let sent = 0; sent < size && !request.closed; sent += chunk.length) {
+				request.write(chunk);
+				await wait(10);
+			}
+			if (!request.closed) {
+				request.end();
+			}
+			await closed;
+			server.off("connection", onConnection);
+			await Promise.all(sockets.map(
+				socket => socket.destroyed || new Promise(resolve => socket.on("close", resolve))
+			));
+			return { status, serverBytesRead: sockets.reduce((total, socket) => total + socket.bytesRead, 0) };
+		}
+		it("should not read the body of a PUT to a stream that doesn't exist", async function() {
+			this.timeout(30000);
+			const size = 64 * 1024 * 1024;
+			const result = await putLargeBody(`${endpoint}/01020304`, size);
+			assert.equal(result.status, 404);
+			assert(result.serverBytesRead < size / 4, `server read ${result.serverBytesRead} of ${size} bytes`);
+		});
+		it("should not read the body of a PUT to a stream that times out", async function() {
+			this.timeout(30000);
+			const size = 64 * 1024 * 1024;
+			const stream = await routes.createProxyStream(controller.app);
+			const result = await putLargeBody(`${endpoint}/${stream.id}`, size);
+			assert.equal(result.status, 500);
+			assert(result.serverBytesRead < size / 4, `server read ${result.serverBytesRead} of ${size} bytes`);
+		});
 		it("should passthrough a stream", async function() {
 			let stream;
 			let responses;