Add graceful closure of HTTP connections

When closing down the HTTP and HTTPS servers existing connections are
not closed by Node.js, and may live on forever if they are actively in
use.  A common case where this happens is if Prometheus is set up with a
short interval for collections.

To close down the active connections they need to be tracked and closed
when the server is closed, add the HttpCloser class which encapsulates
the tracking and closure process behind a simple interface and use this
to close down the HTTP and HTTPS server on the master.

packages/master/master.js

@@ -49,6 +49,7 @@ let db = {
 
 // homebrew modules
 const generateSslCert = require("./src/generate_ssl_cert");
+const HttpCloser = require("./src/HttpCloser");
 const routes = require("./src/routes");
 const libDatabase = require("@clusterio/lib/database");
 const libSchema = require("@clusterio/lib/schema");
@@ -72,7 +73,9 @@ const bodyParser = require("body-parser");
 const fileUpload = require("express-fileupload");
 const app = express();
 let httpServer;
+let httpServerCloser;
 let httpsServer;
+let httpsServerCloser;
 
 app.use(cookieParser());
 app.use(bodyParser.json({
@@ -496,8 +499,8 @@ async function shutdown() {
 
 		let stopTasks = [];
 		logger.info("Stopping HTTP(S) server");
-		if (httpServer) { stopTasks.push(new Promise(resolve => httpServer.close(resolve))); }
-		if (httpsServer) { stopTasks.push(new Promise(resolve => httpsServer.close(resolve))); }
+		if (httpServer) { stopTasks.push(httpServerCloser.close()); }
+		if (httpsServer) { stopTasks.push(httpsServerCloser.close()); }
 		await Promise.all(stopTasks);
 
 		logger.info(`Clusterio cleanly exited in ${Date.now() - exitStartTime}ms`);
@@ -1982,6 +1985,7 @@ async function startServer() {
 	// Only start listening for connections after all plugins have loaded
 	if (httpPort) {
 		httpServer = new (require("http").Server)(app);
+		httpServerCloser = new HttpCloser(httpServer);
 		await listen(httpServer, httpPort, bindAddress);
 		logger.info(`Listening for HTTP on port ${httpServer.address().port}`);
 	}
@@ -2002,6 +2006,7 @@ async function startServer() {
 			key: privateKey,
 			cert: certificate,
 		}, app);
+		httpsServerCloser = new HttpCloser(httpsServer);
 		await listen(httpsServer, httpsPort, bindAddress);
 		logger.info(`Listening for HTTPS on port ${httpsServer.address().port}`);
 	}

packages/master/src/HttpCloser.js

@@ -0,0 +1,109 @@
+"use strict";
+const http = require("http");
+const https = require("https");
+const util = require("util");
+
+/**
+ * Gracefully close a Node.js HTTP(S) server
+ *
+ * Helper class to properly close down a Node.js HTTP(S) server.  This is
+ * necessary due to connections being able to linger around forever with the
+ * keep-alive logic in HTTP.
+ *
+ * @example
+ * // during server initialization
+ * const server = http.createServer();
+ * const serverCloser = new HttpCloser(server);
+ *
+ * // in the shutdown logic
+ * await serverCloser.close();
+ *
+ * @alias module:master/src/HttpCloser
+ */
+class HttpCloser {
+	/**
+	 * Server to attach this closer to.
+	 * @param {https.Server | http.Server} server - HTTP(S) to attach to
+	 */
+	constructor(server) {
+		this._server = server;
+		this._sockets = new Set();
+		this._responses = new Set();
+
+		if (this._server instanceof http.Server) {
+			this._server.on("connection", this._addSocket.bind(this));
+		} else if (this._server instanceof https.Server) {
+			this._server.on("secureConnection", this._addSocket.bind(this));
+		} else {
+			throw Error("server must be an instance of either http.Server or https.Server.");
+		}
+		this._server.on("request", this._addRequest.bind(this));
+	}
+
+	_addSocket(socket) {
+		this._sockets.add(socket);
+		socket.once("close", () => {
+			this._sockets.delete(socket);
+		});
+	}
+
+	_addRequest(request, response) {
+		this._responses.add(response);
+		response.once("close", () => this._responses.delete(response));
+		response.once("finish", () => this._responses.delete(response));
+	}
+
+	/**
+	 * Gracefully close down the attached HTTP(S) server.
+	 *
+	 * Closes open connections that do not have a request in flight, signals
+	 * connections to pending requests should close, and initiates the
+	 * close operation on the attached server.
+	 *
+	 * If there's requests in progress that still haven't been completed
+	 * after the timeout value passed they will be aborted.
+	 *
+	 * @param {number} timeout -
+	 *     Time in ms to wait for connections to close.
+	 */
+	async close(timeout = 5000) {
+		let activeSockets = new Set();
+		for (let response of this._responses) {
+			if (!response.headersSent) {
+				response.setHeader("Connection", "close");
+			} else {
+				let socket = response.socket;
+				response.once("finish", () => {
+					socket.end();
+				});
+			}
+			activeSockets.add(response.socket);
+		}
+		for (let socket of this._sockets) {
+			if (!activeSockets.has(socket)) {
+				socket.end();
+			}
+		}
+
+		return new Promise((resolve, reject) => {
+			let timer = setTimeout(() => {
+				for (let socket of this._sockets) {
+					// Ideally a RST would be sent here to indicate the the
+					// transfer failed, but we can't reliably send that from
+					// Node.js, see nodejs/node#27428.
+					socket.destroy();
+				}
+			}, timeout);
+			this._server.close(err => {
+				clearTimeout(timer);
+				if (err) {
+					reject(err);
+				} else {
+					resolve();
+				}
+			});
+		});
+	}
+};
+
+module.exports = HttpCloser;

test/master/HttpCloser.js

@@ -0,0 +1,236 @@
+"use strict";
+const assert = require("assert").strict;
+const http = require("http");
+const https = require("https");
+const util = require("util");
+const tls = require("tls");
+
+const forge = require("node-forge");
+const phin = require("phin");
+
+const { wait } = require("@clusterio/lib/helpers");
+const HttpCloser = require("@clusterio/master/src/HttpCloser");
+
+// Time to wait during async operations to ensure they happened in order.
+const tick = 20;
+
+const origCreateSecureContext = tls.createSecureContext;
+tls.createSecureContext = function(options) {
+	if (!options.cert || !options.key) {
+		return origCreateSecureContext(options);
+	}
+
+	let lessOptions = { ...options };
+	delete lessOptions.key;
+	delete lessOptions.cert;
+	let ctx = origCreateSecureContext(lessOptions);
+	ctx.context.setCert(options.cert);
+	ctx.context.setKey(options.key, undefined);
+	return ctx;
+};
+
+function serverSuite(proto) {
+	let api = proto === "http" ? http : https;
+	let server;
+	let serverOptions = {};
+	let clientOptions = {};
+	let closer;
+	let isClosed;
+	let addr;
+
+	function setClosed() {
+		isClosed = true;
+	}
+
+	async function get(url, options = {}) {
+		let core = { ...clientOptions, ...(options.core || {}) };
+		return await phin({ ...options, url, core });
+	}
+
+	before(async function() {
+		if (proto === "https") {
+			// Generate TLS certificate
+			let keypair = forge.pki.rsa.generateKeyPair(512);
+			let cert = forge.pki.createCertificate();
+			cert.publicKey = keypair.publicKey;
+			cert.serialNumber = "01";
+			cert.validity.notBefore = new Date();
+			cert.validity.notAfter = new Date(cert.validity.notBefore);
+			cert.validity.notAfter.setFullYear(cert.validity.notAfter.getFullYear() + 1);
+			let attrs = [{ name: "commonName", value: "localhost" }];
+			cert.setSubject(attrs);
+			cert.setIssuer(attrs);
+			cert.sign(keypair.privateKey);
+			serverOptions.key = forge.pki.privateKeyToPem(keypair.privateKey);
+			let pem = forge.pki.certificateToPem(cert);
+			serverOptions.cert = pem;
+			serverOptions.ciphers = "DEFAULT@SECLEVEL=0";
+			clientOptions.ca = pem;
+			clientOptions.ciphers = "DEFAULT@SECLEVEL=0";
+		}
+	});
+	beforeEach(async function() {
+		isClosed = false;
+		server = api.createServer(serverOptions);
+		server.unref();
+		server.on("close", setClosed);
+		closer = new HttpCloser(server);
+		await util.promisify(server.listen.bind(server))();
+		addr = `${proto}://localhost:${server.address().port}/`;
+	});
+
+	afterEach(function() {
+		server.off("close", setClosed);
+	});
+
+	it("should close the server", async function() {
+		await closer.close();
+		assert(isClosed, "Server was not closed");
+	});
+
+	it("should reject with error if the close fails", async function() {
+		await closer.close();
+		await assert.rejects(closer.close(), { code: "ERR_SERVER_NOT_RUNNING" });
+	});
+
+	it("should gracefully let pending requests finish", async function() {
+		server.on("request", (request, response) => {
+			setTimeout(() => { response.end("ok-1"); }, 3*tick);
+		});
+
+		let request = get(addr);
+		request.catch(() => {});
+		await wait(tick);
+		await closer.close();
+
+		let response = await request;
+		assert.equal(response.statusCode, 200, "non-200 response status code");
+		assert.equal(response.body.toString(), "ok-1", "response content does not match expected");
+	});
+
+	it("should set Connection header to close for pending requests", async function() {
+		server.on("request", (request, response) => {
+			setTimeout(() => { response.end("ok-2"); }, 3*tick);
+			assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive");
+		});
+
+		let agent = new api.Agent({ keepAlive: true });
+		let request = get(addr, { core: { agent }});
+		request.catch(() => {});
+		await wait(tick);
+		await closer.close();
+
+		let response = await request;
+		assert.equal(response.statusCode, 200, "non-200 response status code");
+		assert.equal(response.body.toString(), "ok-2", "response content does not match expected");
+		assert.equal(response.headers.connection, "close", "response Connection header not set to close");
+	});
+
+	it("should gracefully close open connections", async function() {
+		server.on("request", (request, response) => {
+			response.end("ok-3");
+			assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive");
+		});
+
+		let agent = new api.Agent({ keepAlive: true });
+		let request = get(addr, { core: { agent }});
+		request.catch(() => {});
+		await wait(tick);
+		await closer.close();
+
+		let response = await request;
+		assert.equal(response.statusCode, 200, "Request replied with non-200 status code");
+		assert.equal(response.body.toString(), "ok-3", "Request content does not match expected");
+	});
+
+	it("should refuse connection if request sent after starting closure", async function() {
+		server.on("request", (request, response) => {
+			setTimeout(() => { response.end("ok-4"); }, 3*tick);
+		});
+
+		let request = get(addr);
+		request.catch(() => {});
+		await wait(tick);
+		let close = closer.close();
+
+		let agent = new api.Agent({ keepAlive: true });
+		await assert.rejects(get(addr, { core: { agent }}), { code: "ECONNREFUSED" });
+
+		let response = await request;
+		assert.equal(response.statusCode, 200, "non-200 response status code");
+		assert.equal(response.body.toString(), "ok-4", "response content does not match expected");
+
+		await close;
+	});
+
+	it("should close connection after in-flight request is over", async function() {
+		server.on("request", (request, response) => {
+			response.write("ok");
+			setTimeout(() => { response.end("-5"); }, 3*tick);
+			assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive");
+		});
+
+		let agent = new api.Agent({ keepAlive: true });
+		let request = get(addr, { core: { agent }});
+		request.catch(() => {});
+		await wait(tick);
+		await closer.close();
+
+		let response = await request;
+		assert.equal(response.statusCode, 200, "non-200 response status code");
+		assert.equal(response.body.toString(), "ok-5", "response content does not match expected");
+	});
+
+	it("should cause connection reset for qeueud request after in-flight request", async function() {
+		server.on("request", (request, response) => {
+			response.write("ok");
+			setTimeout(() => { response.end("-6"); }, 3*tick);
+			assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive");
+		});
+
+		let agent = new api.Agent({ keepAlive: true, maxSockets: 1 });
+		let request = get(addr, { core: { agent }});
+		request.catch(() => {});
+		let droppedRequest = get(addr, { core: { agent }});
+		droppedRequest.catch(() => {});
+		await wait(tick);
+		await closer.close();
+
+		let response = await request;
+		assert.equal(response.statusCode, 200, "non-200 response status code");
+		assert.equal(response.body.toString(), "ok-6", "response content does not match expected");
+		await assert.rejects(droppedRequest, { code: "ECONNRESET" });
+	});
+
+	it("should abort requests that take too long to respond", async function() {
+		server.on("request", (request, response) => {
+			response.write("ok");
+			setTimeout(() => { response.end("-7"); }, 5*tick);
+		});
+
+		let request = get(addr);
+		request.catch(() => {});
+		await wait(tick);
+		await closer.close(tick*2);
+
+		// Due to a bug in phin [https://github.com/ethanent/phin/issues/59]
+		// the request Promise never settles.
+		// TODO: uncomment when bug is fixed in phin.
+		// await assert.rejects(request);
+	});
+}
+
+describe("master/src/HttpCloser.js", function() {
+	describe("class HttpCloser", function() {
+		describe("constructor()", function() {
+			it("should throw if not passed a Server instance", function() {
+				assert.throws(() => new HttpCloser({}));
+			});
+		});
+		for (let proto of ["http", "https"]) {
+			describe(`${proto} server`, function() {
+				serverSuite(proto);
+			});
+		}
+	});
+});