diff --git a/packages/master/master.js b/packages/master/master.js index 14439ada..a0b8af05 100755 --- a/packages/master/master.js +++ b/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}`); } diff --git a/packages/master/src/HttpCloser.js b/packages/master/src/HttpCloser.js new file mode 100644 index 00000000..8a488404 --- /dev/null +++ b/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; diff --git a/test/master/HttpCloser.js b/test/master/HttpCloser.js new file mode 100644 index 00000000..9b59de49 --- /dev/null +++ b/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); + }); + } + }); +});