Merge pull request #351 from Hornwitser/alpha-bugfixes-3

Alpha bugfixes 3

CHANGELOG.md

@@ -53,6 +53,7 @@ Version 2.0.0
 - Added option to auto start instances on slave startup.
 - Added metric mapping of slave and instance ids to their names, allowing
   them to be displayed by name in queries that join with the mapping.
+- Added option to configure timeout for exported Prometheus metrics.
 
 ### Changes
 

docs/configuration.md

@@ -112,6 +112,15 @@ unavoidable.
 Defaults to 30.
 
 
+### master.metrics_timeout
+
+Timeout in seconds before a call to gather metrics from a slave times
+out.  This should be less than both the timeout and collection interval
+configured for the collection job in Prometheus.
+
+Defaults to 8.
+
+
 ### master.default_role_id
 
 Role to automatically grant to new users.  If null no role is granted.

packages/lib/RateLimiter.js

@@ -0,0 +1,104 @@
+"use strict";
+
+
+/**
+ * Helper class for rate limiting actions
+ *
+ * Provides a simple interface for implementing actions that are rate
+ * limited on invocations.
+ *
+ * @example
+ * const rateLimiter = RateLimiter({
+ *     action: () => { console.log("action"); },
+ *     maxRate: 1,
+ * });
+ *
+ * for (let i = 0; i < 100; i++) {
+ *     rateLimiter.activate();
+ *     // Prints action twice to the console, once for the first activation
+ *     // and one more time 1 second later for 99 other activations that were
+ *     // exceeding the rate limit.
+ * }
+ * @alias module:lib/RateLimiter
+ */
+class RateLimiter {
+	/**
+	 * Construct RateLimiter helper
+	 *
+	 * The options passed may also be modified on the instance at runtime.
+	 *
+	 * @param {object=} options - Options for this rate limiter
+	 * @param {number=} options.maxRate -
+	 *     Maximum rate of activations per second.
+	 * @param {function()=} options.action -
+	 *     Callback invoked at maste `maxRate` per second on activation.
+	 */
+	constructor(options = {}) {
+		/*
+		 * Maximum rate of activations per second.
+		 * @type {number}
+		 */
+		this.maxRate = options.maxRate || 1;
+
+		/**
+		 * Callback invoked at most `maxRate` per second on activation.
+		 * @type {function()}
+		 */
+		this.action = options.action || null;
+
+		this._lastRun = 0;
+		this._runTimeout = null;
+		this.lastProgressBroadcast = Date.now();
+	}
+
+	/**
+	 * Activate the rate limited action
+	 *
+	 * Checks if the invocation of the rate limited action now would fall
+	 * within the rate limit.  If it is within the limit then the action is
+	 * invoked immediatly, otherwise a timeout is registered to invoke the
+	 * action at next point in time where it will be within the rate limit.
+	 *
+	 * @returns {boolean} true if activation was within the rate limit.
+	 */
+	activate() {
+		let now = Date.now();
+		if (now < this._lastRun + 1000 / this.maxRate) {
+			if (!this._runTimeout && this.action !== null) {
+				this._runTimeout = setTimeout(() => {
+					this._runTimeout = null;
+					this._lastRun = Date.now();
+					this.action();
+				}, this._lastRun + 1000 / this.maxRate - now);
+			}
+			return false;
+		}
+
+		// We may end up activating before the timeout does.
+		if (this._runTimeout) {
+			clearTimeout(this._runTimeout);
+			this._runTimeout = null;
+		}
+
+		this._lastRun = now;
+		if (this.action !== null) {
+			this.action();
+		}
+		return true;
+	}
+
+	/**
+	 * Cancel pending action
+	 *
+	 * Cancel pending action invocation.  This is necessary when shutting
+	 * down the use of the rate limit as a timeout is registered to invoke
+	 * the next action when activated faster than the rate limit allows.
+	 */
+	cancel() {
+		if (this._runTimeout) {
+			clearTimeout(this._runTimeout);
+		}
+	}
+}
+
+module.exports = RateLimiter;

packages/lib/config/definitions.js

@@ -94,6 +94,13 @@ MasterGroup.define({
 	type: "number",
 	initial_value: 30,
 });
+MasterGroup.define({
+	name: "metrics_timeout",
+	title: "Metrics Timeout",
+	description: "Timeout in seconds for metrics gathering from slaves",
+	type: "number",
+	initial_value: 8,
+});
 MasterGroup.define({
 	name: "default_role_id",
 	title: "Default role",

packages/lib/factorio/server.js

@@ -522,6 +522,8 @@ class FactorioServer extends events.EventEmitter {
 	 * @param {number} options.maxConcurrentCommands -
 	 *     Maximum number of RCON commands transmitted in parallel on the
 	 *     RCON connection.
+	 * @param {number} options.hangTimeout -
+	 *     Timeout in ms to wait for a response to a shutdown command.
 	 */
 	constructor(factorioDir, writeDir, options) {
 		super();
@@ -547,6 +549,8 @@ class FactorioServer extends events.EventEmitter {
 		this.verboseLogging = options.verboseLogging;
 		/** Maximum number of RCON commands transmitted in parallel on the RCON connection  */
 		this.maxConcurrentCommands = options.maxConcurrentCommands || 5;
+		/** Timeout in ms to wait for a response to a shutdown command */
+		this.hangTimeout = options.hangTimeout || 5000;
 		this._state = "new";
 		this._server = null;
 		this._rconClient = null;
@@ -637,7 +641,7 @@ class FactorioServer extends events.EventEmitter {
 		}
 
 		if (!this.emit(`ipc-${channel}`, content)) {
-			this.logger.warn(`Warning: Unhandled ipc-${channel}`, { content });
+			this._logger.warn(`Warning: Unhandled ipc-${channel}`, { content });
 		}
 	}
 
@@ -704,8 +708,6 @@ class FactorioServer extends events.EventEmitter {
 		this._rconClient.on("authenticated", () => { this._rconReady = true; this.emit("rcon-ready"); });
 		this._rconClient.on("end", () => {
 			this._rconClient = null;
-			this._rconReady = false;
-			this._startRcon().catch(() => { /* Ignore */ });
 		});
 		// XXX: Workaround to suppress bogus event listener warning
 		if (this._rconClient.emitter instanceof events.EventEmitter) {
@@ -802,6 +804,10 @@ class FactorioServer extends events.EventEmitter {
 				}
 			}
 
+			if (this._rconClient) {
+				this._rconClient.end().catch(() => {});
+			}
+
 			// Reset server state
 			this._state = "init";
 			this._server = null;
@@ -954,6 +960,9 @@ class FactorioServer extends events.EventEmitter {
 		if (!this._rconReady) {
 			await events.once(this, "rcon-ready");
 		}
+		if (!this._rconClient) {
+			throw new Error("RCON connection lost");
+		}
 
 		let response = await this._rconClient.send(message);
 		if (expectEmpty && response !== "") {
@@ -972,17 +981,25 @@ class FactorioServer extends events.EventEmitter {
 		this._check(["running"]);
 
 		let hanged = true;
+		let stopped = false;
 		function setAlive() { hanged = false; }
 
 		let timeoutId = setTimeout(() => {
 			if (hanged) {
-				logger.error("Factorio appears to have hanged, sending SIGKILL");
+				this._logger.error("Factorio appears to have hanged, sending SIGKILL");
 				if (this._rconClient) {
 					this._rconClient.end().catch(() => {});
 				}
 				this._server.kill("SIGKILL");
 			}
-		}, 5000);
+		}, this.hangTimeout);
+
+		// It's possible the server decides to exit on its own, in which
+		// case the stop operation has to be cancelled.
+		this._server.once("exit", () => {
+			stopped = true;
+			clearTimeout(timeoutId);
+		});
 
 		this._state = "stopping";
 		// On Windows we need to save the map as there's no graceful shutdown.
@@ -1014,6 +1031,12 @@ class FactorioServer extends events.EventEmitter {
 			await this._rconClient.end();
 		}
 
+		// The Factorio server may have decided to get ahead of us and
+		// stop by itself before we had the chance to signal it.
+		if (stopped) {
+			return;
+		}
+
 		// On linux this sends SIGTERM to the process.  On windows this
 		// will terminate the process with no cleanup.  You can send your
 		// complaint for the lack of graceful shutdown to the Factorio

packages/lib/link/messages.js

@@ -207,12 +207,13 @@ class Request extends Message {
 						response = {};
 					}
 
-					if (!this._responseValidator({ seq: 0, type: this.responseType, response })) {
+					let data = { ...response, seq: message.seq };
+					if (!this._responseValidator({ seq: 0, type: this.responseType, data })) {
 						logger.error(JSON.stringify(this._responseValidator.errors, null, 4));
 						throw new Error(`Validation failed responding to ${this.requestType}`);
 					}
 
-					link.connector.send(this.responseType, { ...response, seq: message.seq });
+					link.connector.send(this.responseType, data);
 
 				}).catch(err => {
 					if (!(err instanceof libErrors.RequestError)) {

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({
@@ -190,8 +193,9 @@ async function getMetrics(req, res, next) {
 	}
 
 	let requests = [];
+	let timeout = masterConfig.get("master.metrics_timeout") * 1000;
 	for (let slaveConnection of slaveConnections.values()) {
-		requests.push(libHelpers.timeout(libLink.messages.getMetrics.send(slaveConnection), 8e3, null));
+		requests.push(libHelpers.timeout(libLink.messages.getMetrics.send(slaveConnection), timeout, null));
 	}
 
 	for await (let result of await libPrometheus.defaultRegistry.collect()) {
@@ -495,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`);
@@ -1981,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}`);
 	}
@@ -2001,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;

packages/slave/slave.js

@@ -1647,6 +1647,14 @@ I           version of clusterio.  Expect things to break. I
 		if (err instanceof libErrors.AuthenticationFailed) {
 			logger.fatal(err.message);
 
+		} else if (err instanceof libErrors.PluginError) {
+			logger.fatal(`
+${err.pluginName} plugin threw an unexpected error
+during startup, please report it to the plugin author.
+------------------------------------------------------
+${err.original.stack}`
+			);
+
 		} else {
 			logger.fatal(`
 +--------------------------------------------------------------+

plugins/research_sync/master.js

@@ -3,41 +3,27 @@ const fs = require("fs-extra");
 const path = require("path");
 
 const libPlugin = require("@clusterio/lib/plugin");
+const RateLimiter = require("@clusterio/lib/RateLimiter");
 
 
 class MasterPlugin extends libPlugin.BaseMasterPlugin {
 	async init() {
 		this.technologies = await loadTechnologies(this.master.config, this.logger);
+		this.progressRateLimiter = new RateLimiter({
+			maxRate: 1,
+			action: () => this.broadcastProgress(),
+		});
+
 		this.lastProgressBroadcast = Date.now();
 		this.progressBroadcastId = null;
 		this.progressToBroadcast = new Set();
 	}
 
 	async onShutdown() {
-		if (this.progressBroadcastId) {
-			clearTimeout(this.progressBroadcastId);
-		}
+		this.progressRateLimiter.cancel();
 		await saveTechnologies(this.master.config, this.technologies, this.logger);
 	}
 
-	registerProgress() {
-		// Rate limit progress broadcasts to one per second
-		if (Date.now() < this.lastProgressBroadcast + 1000) {
-			if (!this.progressBroadcastId) {
-				this.progressBroadcastId = setTimeout(() => {
-					this.broadcastProgress();
-					this.progressBroadcastId = null;
-				}, 2000);
-			}
-
-		} else {
-			if (this.progressBroadcastId) {
-				clearTimeout(this.progressBroadcastId);
-			}
-			this.broadcastProgress();
-		}
-	}
-
 	broadcastProgress() {
 		let techs = [];
 		for (let name of this.progressToBroadcast) {
@@ -81,7 +67,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin {
 		if (newProgress < 1) {
 			tech.progress = newProgress;
 			this.progressToBroadcast.add(name);
-			this.registerProgress();
+			this.progressRateLimiter.activate();
 
 		} else {
 			tech.researched = true;
@@ -148,7 +134,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin {
 				}
 			}
 		}
-		this.registerProgress();
+		this.progressRateLimiter.activate();
 
 		let technologies = [];
 		for (let [name, tech] of this.technologies) {

plugins/statistics_exporter/README.md

@@ -7,7 +7,7 @@ Clusterio.
 
 
 Installation
-============
+------------
 
 Run the following commands in the folder Clusterio is installed to:
 
@@ -16,3 +16,18 @@ Run the following commands in the folder Clusterio is installed to:
 
 Substitute clusteriomaster with clusterioslave or clusterioctl if this a
 dedicate slave or ctl installation respectively.
+
+
+Instance Configuration
+----------------------
+
+### statistics_exporter.command_timeout
+
+Timeout in seconds to wait for the statitics gathering command to the
+Factorio server return the result.  If the timeout is exceeded the
+previous collected results are returned instead.  Normally the command
+returns immediatly with the data, but if the RCON interface is
+overloaded then the command may take too long to timely answer the
+metrics collection request.
+
+Defaults to `1`.

plugins/statistics_exporter/info.js

@@ -1,8 +1,21 @@
 "use strict";
+const libConfig = require("@clusterio/lib/config");
+
+class InstanceConfigGroup extends libConfig.PluginConfigGroup {}
+InstanceConfigGroup.groupName = "statistics_exporter";
+InstanceConfigGroup.define({
+	name: "command_timeout",
+	title: "Command Timeout",
+	description: "Timeout in seconds of the metrics command before returning results from the previous invocation.",
+	type: "number",
+	initial_value: 1,
+});
+InstanceConfigGroup.finalize();
 
 module.exports = {
 	name: "statistics_exporter",
 	title: "Prometheus Statistics Export",
 	description: "Exports statistics to Prometheus",
 	instanceEntrypoint: "instance",
+	InstanceConfigGroup,
 };

plugins/statistics_exporter/instance.js

@@ -73,11 +73,12 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin {
 	}
 
 	async onMetrics() {
-		// Wait up to 5 seconds for the metrics to be collected.  It may
+		// Wait configured timeout for the metrics to be collected.  It may
 		// take a long time for the command to go through if the command
-		// stream is overloaded.  Should this take more than 5 seconds the
+		// stream is overloaded.  Should the timeout be exceeded the
 		// previous values for the metrics will end up being sent to master.
-		await libHelpers.timeout(this.gatherMetrics(), 5e3);
+		let timeout = this.instance.config.get("statistics_exporter.command_timeout") * 1000;
+		await libHelpers.timeout(this.gatherMetrics(), timeout);
 	}
 }
 

plugins/statistics_exporter/test/plugin.js

@@ -9,8 +9,10 @@ const info = require("../info");
 describe("statistics_exporter plugin", function() {
 	describe("class InstancePlugin", function() {
 		let instancePlugin;
+		let mockInstance = new mock.MockInstance();
+		mockInstance.mockConfigEntries.set("statistics_exporter.command_timeout", 1);
 		before(async function() {
-			instancePlugin = new instance.InstancePlugin(info, new mock.MockInstance());
+			instancePlugin = new instance.InstancePlugin(info, mockInstance);
 			await instancePlugin.init();
 		});
 

test/lib/RateLimiter.js

@@ -0,0 +1,64 @@
+"use strict";
+const assert = require("assert").strict;
+
+const { wait } = require("@clusterio/lib/helpers");
+const RateLimiter = require("@clusterio/lib/RateLimiter");
+
+
+describe("lib/RateLimiter", function() {
+	describe("class RateLimiter", function() {
+		let rateLimiter;
+		beforeEach(function() {
+			rateLimiter = new RateLimiter();
+		});
+		afterEach(function() {
+			rateLimiter.cancel();
+		});
+
+		it("should rate limit activations", function() {
+			assert(rateLimiter.activate(), "fist invocation was rate limited");
+			assert(!rateLimiter.activate(), "second invocation was not rate limited");
+		});
+
+		it("should invoke action immediately when not rate limited", function() {
+			let invoked = false;
+			rateLimiter.action = () => { invoked = true; };
+			rateLimiter.activate();
+			assert(invoked, "action was not invoked");
+		});
+
+		it("should invoke action at a later point in time if rate limited", async function() {
+			let invokedCount = 0;
+			rateLimiter.maxRate = 100;
+			rateLimiter.action = () => { invokedCount += 1; };
+			rateLimiter.activate();
+			rateLimiter.activate();
+			assert.equal(invokedCount, 1, "expected one invocation");
+			await wait(20);
+			assert.equal(invokedCount, 2, "expected two invocations");
+		});
+
+		it("should invoke action before timeout if maxRate changed", async function() {
+			let invokedCount = 0;
+			rateLimiter.action = () => { invokedCount += 1; };
+			rateLimiter.activate();
+			rateLimiter.activate();
+			assert.equal(invokedCount, 1, "expected one invocation");
+			await wait(20);
+			rateLimiter.maxRate = 100;
+			rateLimiter.activate();
+			assert.equal(invokedCount, 2, "expected two invocations");
+		});
+
+		it("should cancel action if canceled before timeout", async function() {
+			let invokedCount = 0;
+			rateLimiter.action = () => { invokedCount += 1; };
+			rateLimiter.activate();
+			rateLimiter.activate();
+			rateLimiter.cancel();
+			assert.equal(invokedCount, 1, "expected one invocation");
+			await wait(20);
+			assert.equal(invokedCount, 1, "expected one invocation");
+		});
+	});
+});

test/lib/factorio/server.js

@@ -5,6 +5,7 @@ const fs = require("fs-extra");
 const path = require("path");
 
 const libFactorio = require("@clusterio/lib/factorio");
+const { wait } = require("@clusterio/lib/helpers");
 const { testLines } = require("./lines");
 
 
@@ -243,5 +244,31 @@ describe("lib/factorio/server", function() {
 				assert(!await fs.pathExists(filePath), "File was not deleted");
 			});
 		});
+
+		describe(".stop()", function() {
+			it("should handle server quitting on its own during stop", async function() {
+				if (process.platform === "win32") {
+					this.skip();
+				}
+				server.hangTimeout = 20;
+				server._server = new events.EventEmitter();
+				server._server.kill = () => true;
+				server._state = "running";
+				server._rconReady = true;
+				server._rconClient = {
+					async end() {
+						server._rconClient = null;
+						process.nextTick(() => {
+							server.emit("_quitting");
+							server._server.emit("exit");
+						});
+					},
+				};
+				server._watchExit();
+
+				await server.stop();
+				await wait(21); // Wait until after hang timeout
+			});
+		});
 	});
 });

test/lib/link/messages.js

@@ -20,6 +20,9 @@ describe("lib/link/messages", function() {
 			requestProperties: {
 				"test": { type: "string" },
 			},
+			responseProperties: {
+				"test": { type: "string" },
+			},
 		});
 
 		describe("constructor", function() {
@@ -63,7 +66,24 @@ describe("lib/link/messages", function() {
 				});
 			});
 			it("should implicitly send an empty object on empty return", async function() {
-				handlerResult = undefined;
+				let emptyRequest = new libLink.Request({
+					type: "empty",
+					links: ["source-target"],
+				});
+				emptyRequest.attach(testTargetLink, async(message) => {});
+				testTargetLink._handlers.get("empty_request")({ seq: 2 });
+				let result = await new Promise(resolve => {
+					testTargetLink.connector.send = (type, data) => resolve({ type, data });
+				});
+				assert.deepEqual(result, {
+					type: "empty_response",
+					data: {
+						seq: 2,
+					},
+				});
+			});
+			it("should validate response from handler", async function() {
+				handlerResult = { test: 12 };
 				testTargetLink._handlers.get("test_request")({ seq: 2 });
 				let result = await new Promise(resolve => {
 					testTargetLink.connector.send = (type, data) => resolve({ type, data });
@@ -71,6 +91,7 @@ describe("lib/link/messages", function() {
 				assert.deepEqual(result, {
 					type: "test_response",
 					data: {
+						error: "Validation failed responding to test_request",
 						seq: 2,
 					},
 				});

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);
+			});
+		}
+	});
+});

test/mock.js

@@ -103,11 +103,16 @@ class MockInstance extends libLink.Link {
 		this.logger = new MockLogger();
 		this.server = new MockServer();
 		this.name = "test";
+		this.mockConfigEntries = new Map([
+			["instance.id", 7357],
+			["factorio.enable_save_patching", true],
+		]);
 		this.config = {
 			get: (name) => {
-				if (name === "instance.id") { return 7357; }
-				if (name === "factorio.enable_save_patching") { return true; }
-				throw Error("Not implemented");
+				if (this.mockConfigEntries.has(name)) {
+					return this.mockConfigEntries.get(name);
+				}
+				throw Error(`mock for field ${name} is not implemented`);
 			},
 		};
 	}