Merge pull request #412 from Hornwitser/small-things-5

Small things 5

README.md

@@ -115,8 +115,8 @@ Otherwise see below for OS specific instructions.
 
 3.  If you chose to use local factorio directory for the Factorio installation then download the headless build of Factorio and unpack it:
 
-        wget -O factorio.tar.gz https://www.factorio.com/get-download/latest/headless/linux64
-        tar -xf factorio.tar.gz
+        wget -O factorio.tar.xz https://www.factorio.com/get-download/latest/headless/linux64
+        tar -xf factorio.tar.xz
 
     To specify a version of Factorio to download replace "latest" in the URL with a version number like "1.0.0".
 

docs/writing-plugins.md

@@ -502,11 +502,11 @@ This is signaled to `MasterPlugin` classes via the `onPrepareSlaveDisconnect` ho
 
 After the prepare disconnect the connection will be closed, which will result in pending requests and events being dropped.
 Plugins must respond to the prepare disconnect by stopping any processess it does that send events or requests over the link in question.
-This can be accomplished either through listening for the prepare disconnect hook, or by checking the `connected` or `closing` property of the connector for the master/slave connection.
-For example the sending of an event from an `InstancePlugin` class can be stopped while the connection is closing by using the following code:
+This can be accomplished either through listening for the prepare disconnect hook, or by checking the `connected` property of the `SlaveConnection` class and `Slave` class on the master and slave respectively.
+For example the sending of an event from an `InstancePlugin` class can be stopped while the connection is not connected, not in the dropped state, and not in the process of discunnecting by using the following code:
 
 ```js
-if (!this.slave.connector.closing) {
+if (this.slave.connected) {
     this.info.messages.frobnicate.send(this.instance, { foo: "bar" });
 }
 ```
@@ -517,8 +517,8 @@ The re-establishement of the connection is  notified to plugins via the `connect
 The second connection event which is of lesser importance to respond to is the `drop` connection event served through `onMasterConnectionEvent` for `InstancePlugin` classes and through `onSlaveConnectionEvent` for `MasterPlugin` classes.
 This is raised when the connection between the master and slave in question is lost, most likely due to networking issues.
 When in the dropped state the slave will keep trying to reconnect to the master server in order to re-establish it, and if successful no events or requests will be lost.
-However while in the dropped state any requests and events sent gets queued up in memory until the connection is re-established.
-This means that if your plugin sends a lot of events or requests, they can end up being queued up in a buffer for a long time and sent out all at once.
+However while in the dropped state any requests and events sent gets queued up in memory until the connection is either re-established or the session times out.
+This means that if your plugin sends a lot of events or requests, they can end up being queued up in a buffer and sent out all at once the connection is re-estabished.
 To avoid this you should be throtteling and/or stopping your requests/events after `drop` has been raised, and continue back as normal when `resume` is raised.
 
 

packages/create/create.js

@@ -19,6 +19,7 @@ const scriptExt = process.platform === "win32" ? ".cmd" : "";
 
 const availablePlugins = [
 	{ name: "Global Chat", value: "@clusterio/plugin-global_chat" },
+	{ name: "Inventory Sync", value: "@clusterio/plugin-inventory_sync" },
 	{ name: "Player Auth", value: "@clusterio/plugin-player_auth" },
 	{ name: "Research Sync", value: "@clusterio/plugin-research_sync" },
 	{ name: "Statistics Exporter", value: "@clusterio/plugin-statistics_exporter" },

packages/ctl/ctl.js

@@ -819,6 +819,8 @@ logCommands.add(new libCommand.Command({
 			"slave": { describe: "Query log of given slave", nargs: 1, type: "string", default: null },
 			"instance": { describe: "Query log of given instance", nargs: 1, type: "string", default: null },
 			"max-level": { describe: "Maximum log level to return", nargs: 1, type: "string", default: null },
+			"limit": { describe: "Max number of entries to return", nargs: 1, type: "number", default: 1000 },
+			"start": { describe: "Limit from the start instead of the end", nargs: 0, type: "boolean", default: false },
 		});
 	}],
 	handler: async function(args, control) {
@@ -835,8 +837,14 @@ logCommands.add(new libCommand.Command({
 			slave_ids,
 			instance_ids,
 			max_level: args.maxLevel,
+			limit: args.limit,
+			order: args.start ? "asc" : "desc",
 		});
 
+		if (!args.start) {
+			result.log.reverse();
+		}
+
 		let stdoutLogger = winston.createLogger({
 			level: "verbose",
 			levels,

packages/lib/link/connectors.js

@@ -163,6 +163,14 @@ class WebSocketBaseConnector extends events.EventEmitter {
 	get closing() {
 		return this._closing || this._state === "closed";
 	}
+
+	/**
+	 * True if the connector currently has a valid session that is either
+	 * connected or in the process of resuming.
+	 */
+	get hasSession() {
+		return ["connected", "resuming"].includes(this._state);
+	}
 }
 
 /**

packages/lib/link/messages.js

@@ -859,6 +859,8 @@ messages.queryLog = new Request({
 			items: { type: "integer" },
 		},
 		"max_level": { type: ["string", "null"] },
+		"limit": { type: "integer" },
+		"order": { enum: ["asc", "desc"] },
 	},
 	responseProperties: {
 		"log": {

packages/lib/plugin.js

@@ -256,6 +256,7 @@ class BaseMasterPlugin {
 	 * or received notice from a slave that the status of an instance has
 	 * changed.  The possible statuses that can be notified about are:
 	 * - `unassigned:`: Instance is no longer asssigned to a slave.
+	 * - `unknown`: Slave assigned to instance is offline.
 	 * - `stopped`: Instance is no longer running or was just assigned to a
 	 *   slave.
 	 * - `starting`: Instance is in the process of starting up.
@@ -270,7 +271,8 @@ class BaseMasterPlugin {
 	 * they are assigned to a slave, when the slave then connects to the
 	 * master and informs of the current status of its instances this hook
 	 * is invoked for all those instances.  You can detect this situation by
-	 * checking if prev equals `unknown`.
+	 * checking if prev equals `unknown`.  If the slave disconnects the
+	 * instances will again get the `unknown` status.
 	 *
 	 * When instances are created on the master they will notify of a status
 	 * change with prev set to null.  While the status of new instances is

packages/master/src/BaseConnection.js

@@ -16,6 +16,7 @@ class BaseConnection extends libLink.Link {
 		/** @member {module:master/src/WsServerConnector} module:master/src/BaseConnection#connector */
 		super("master", target, connector);
 		this._master = master;
+		this._disconnecting = false;
 		libLink.attachAllMessages(this);
 		for (let masterPlugin of this._master.plugins.values()) {
 			libPlugin.attachPluginMessages(this, masterPlugin);
@@ -57,6 +58,7 @@ class BaseConnection extends libLink.Link {
 
 	async prepareDisconnectRequestHandler(message, request) {
 		await libPlugin.invokeHook(this._master.plugins, "onPrepareSlaveDisconnect", this);
+		this._disconnecting = true;
 		this.connector.setClosing();
 		return await super.prepareDisconnectRequestHandler(message, request);
 	}
@@ -72,6 +74,15 @@ class BaseConnection extends libLink.Link {
 
 		await this.connector.close(code, reason);
 	}
+
+	/**
+	 * True if the link is connected, not in the dropped state and not in
+	 * the process of disconnecting.
+	 * @type {boolean}
+	 */
+	get connected() {
+		return !this._disconnecting && this.connector.connected;
+	}
 }
 
 module.exports = BaseConnection;

packages/master/src/ControlConnection.js

@@ -456,12 +456,15 @@ class ControlConnection extends BaseConnection {
 		// - The log file is read starting from the beginning
 		let setDuration = lastQueryLogTime.startTimer();
 		let log = await query({
-			order: "asc",
+			order: message.data.order,
 			limit: Infinity,
 			from: new Date(0),
 		});
 
 		log = log.filter(this.constructor.logFilter(message.data));
+		if (message.data.limit) {
+			log = log.slice(0, message.data.limit);
+		}
 		setDuration();
 		return { log };
 	}

packages/master/src/SlaveConnection.js

@@ -39,6 +39,20 @@ class SlaveConnection extends BaseConnection {
 				}
 			});
 		}
+
+		this.connector.on("close", () => {
+			// Update status to unknown for instances on this slave.
+			for (let instance of this._master.instances.values()) {
+				if (instance.config.get("instance.assigned_slave") !== this._id) {
+					continue;
+				}
+
+				let prev = instance.status;
+				instance.status = "unknown";
+				this._master.instanceUpdated(instance);
+				libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, prev);
+			}
+		});
 	}
 
 	/**

packages/master/src/WsServer.js

@@ -258,7 +258,7 @@ ${err.stack}`
 		let connector = new WsServerConnector(sessionId, sessionTimeout, heartbeatInterval);
 		this.activeConnectors.set(sessionId, connector);
 		connector.on("close", () => {
-			this.activeConnectors.delete(connector);
+			this.activeConnectors.delete(sessionId);
 		});
 
 		let additionalReadyData = {};

packages/slave/modules/clusterio/serialize.lua

@@ -11,7 +11,7 @@ local function version_to_table(version)
 end
 
 -- 0.17 compatibility
-local supports_bar, get_bar, has_bar, version
+local supports_bar, get_bar, set_bar, version
 if (pcall(function() local mods = script.active_mods end)) then
 	supports_bar = "supports_bar"
 	get_bar = "get_bar"

packages/slave/slave.js

@@ -295,10 +295,6 @@ class Instance extends libLink.Link {
 		for (let pluginInstance of this.plugins.values()) {
 			pluginInstance.onExit();
 		}
-
-		this.sendSaveListUpdate().catch(err => {
-			this.logger.error(`Unexpected error updating save list:\n${err.stack}`);
-		});
 	}
 
 	async _loadPlugin(pluginInfo, slave) {
@@ -318,6 +314,7 @@ class Instance extends libLink.Link {
 			await this.server.init();
 		} catch (err) {
 			this.notifyExit();
+			await this.sendSaveListUpdate();
 			throw err;
 		}
 
@@ -335,6 +332,7 @@ class Instance extends libLink.Link {
 				await this._loadPlugin(pluginInfo, this._slave);
 			} catch (err) {
 				this.notifyExit();
+				await this.sendSaveListUpdate();
 				throw err;
 			}
 		}
@@ -689,6 +687,7 @@ class Instance extends libLink.Link {
 		if (this.server._state === "running") {
 			await libPlugin.invokeHook(this.plugins, "onStop");
 			await this.server.stop();
+			await this.sendSaveListUpdate();
 		}
 	}
 
@@ -728,6 +727,7 @@ class Instance extends libLink.Link {
 			saveName = await this.prepareSave(saveName);
 		} catch (err) {
 			this.notifyExit();
+			await this.sendSaveListUpdate();
 			throw err;
 		}
 
@@ -749,6 +749,7 @@ class Instance extends libLink.Link {
 			await this.prepare();
 		} catch (err) {
 			this.notifyExit();
+			await this.sendSaveListUpdate();
 			throw err;
 		}
 
@@ -779,12 +780,14 @@ class Instance extends libLink.Link {
 
 		} catch (err) {
 			this.notifyExit();
+			await this.sendSaveListUpdate();
 			throw err;
 		}
 
 		this.server.on("exit", () => this.notifyExit());
 		let { name, seed, map_gen_settings, map_settings } = message.data;
 		await this.server.create(name, seed, map_gen_settings, map_settings);
+		await this.sendSaveListUpdate();
 		this.logger.info("Successfully created save");
 	}
 
@@ -942,6 +945,10 @@ class InstanceConnection extends libLink.Link {
 	}
 
 	async forwardEventToMaster(message, event) {
+		if (!this.slave.connector.hasSession) {
+			return;
+		}
+
 		event.send(this.slave, message.data);
 	}
 
@@ -1582,10 +1589,10 @@ class Slave extends libLink.Link {
 	}
 
 	async prepareDisconnectRequestHandler(message, request) {
-		this._disconnecting = true;
 		for (let instanceConnection of this.instanceConnections.values()) {
 			await libLink.messages.prepareMasterDisconnect.send(instanceConnection);
 		}
+		this._disconnecting = true;
 		this.connector.setClosing();
 		return await super.prepareDisconnectRequestHandler(message, request);
 	}
@@ -1628,6 +1635,15 @@ ${err.stack}`
 			process.exit(1);
 		}
 	}
+
+	/**
+	 * True if the connection to the master is connected, not in the dropped
+	 * state,and not in the process of disconnecting.
+	 * @type {boolean}
+	 */
+	get connected() {
+		return !this._disconnecting && this.connector.connected;
+	}
 }
 
 /**

packages/web_ui/src/components/InstanceConsole.jsx

@@ -65,8 +65,10 @@ export default function InstanceConsole(props) {
 			slave_ids: [],
 			instance_ids: [props.id],
 			max_level: null,
+			limit: 400,
+			order: "desc",
 		}).then(result => {
-			setPastLines(result.log.slice(-400).map((info, index) => formatLog(info, index)));
+			setPastLines(result.log.map((info, index) => formatLog(info, index)).reverse());
 		}).catch(err => {
 			setPastLines([<span key={0}>{`Error loading log: ${err.message}`}<br/></span>]);
 		});

packages/web_ui/src/components/SavesList.jsx

@@ -29,6 +29,7 @@ export default function SavesList(props) {
 	let [starting, setStarting] = useState(false);
 	let [uploadingFiles, setUploadingFiles] = useState([]);
 
+	let slaveOffline = ["unassigned", "unknown"].includes(props.instance.status);
 	const saveTable = <Table
 		size="small"
 		columns={[
@@ -79,6 +80,7 @@ export default function SavesList(props) {
 					}}
 				>Load save</Button>
 				<Button
+					disabled={slaveOffline}
 					onClick={() => {
 						libLink.messages.downloadSave.send(
 							control, { instance_id: props.instance.id, save: save.name }
@@ -102,7 +104,7 @@ export default function SavesList(props) {
 						).catch(notifyErrorHandler("Error deleting save"));
 					}}
 				>
-					<Button danger>Delete</Button>
+					<Button danger disabled={slaveOffline}>Delete</Button>
 				</Popconfirm>
 			</Space>,
 		}}
@@ -141,6 +143,7 @@ export default function SavesList(props) {
 	}
 
 	let uploadProps = {
+		disabled: slaveOffline,
 		name: "file",
 		accept: ".zip",
 		headers: {
@@ -157,7 +160,7 @@ export default function SavesList(props) {
 	return <div>
 		<SectionHeader title="Saves" extra=<Space>
 			<Upload {...uploadProps} >
-				<Button>Upload save</Button>
+				<Button disabled={slaveOffline}>Upload save</Button>
 			</Upload>
 			<CreateSaveModal instance={props.instance} />
 		</Space> />

packages/web_ui/src/util/websocket.jsx

@@ -133,7 +133,7 @@ export class Control extends libLink.Link {
 			throw new Error(`Given handler is not registered for slave ${id}`);
 		}
 
-		handlers.slice(index, 1);
+		handlers.splice(index, 1);
 		if (!handlers.length) {
 			this.slaveUpdateHandlers.delete(id);
 			await this.updateSlaveSubscriptions();
@@ -242,12 +242,12 @@ export class Control extends libLink.Link {
 		let index = handlers.lastIndexOf(handler);
 		if (index === -1) {
 			throw new Error(`Given handler is not registered for instance ${id}`);
+		}
 
-			handlers.splice(index, 1);
-			if (!handlers.length) {
-				this.saveListUpdateHandlers.delete(id);
-				await this.updateSaveListSubscriptions();
-			}
+		handlers.splice(index, 1);
+		if (!handlers.length) {
+			this.saveListUpdateHandlers.delete(id);
+			await this.updateSaveListSubscriptions();
 		}
 	}
 

plugins/player_auth/module/auth.lua

@@ -1,7 +1,7 @@
 local clusterio_api = require("modules/clusterio/api")
 local auth = {}
 
-function open_dialog(player, url, code)
+local function open_dialog(player, url, code)
 	if player.gui.screen.player_auth_dialog then
 		player.gui.screen.player_auth_dialog.destroy()
 	end

plugins/subspace_storage/instance.js

@@ -10,13 +10,12 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin {
 	}
 
 	async init() {
-		this.stopping = false;
 		this.pendingTasks = new Set();
 		this.instance.server.on("ipc-subspace_storage:output", (output) => {
 			this.provideItems(output).catch(err => this.unexpectedError(err));
 		});
 		this.instance.server.on("ipc-subspace_storage:orders", (orders) => {
-			if (this.stopping) {
+			if (this.instance.status !== "running" || !this.slave.connected) {
 				return;
 			}
 
@@ -28,6 +27,9 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin {
 
 	async onStart() {
 		this.pingId = setInterval(() => {
+			if (!this.slave.connected) {
+				return; // Only ping if we are actually connected to the master.
+			}
 			this.sendRcon(
 				"/sc __subspace_storage__ global.ticksSinceMasterPinged = 0", true
 			).catch(err => this.unexpectedError(err));
@@ -41,7 +43,6 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin {
 
 	async onStop() {
 		clearInterval(this.pingId);
-		this.stopping = true;
 		await Promise.all(this.pendingTasks);
 	}
 
@@ -51,6 +52,16 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin {
 
 	// provide items --------------------------------------------------------------
 	async provideItems(items) {
+		if (!this.slave.connector.hasSession) {
+			// For now the items are voided if the master connection is
+			// down, which is no different from the previous behaviour.
+			if (this.instance.config.get("subspace_storage.log_item_transfers")) {
+				this.logger.verbose("Voided the following items:");
+				this.logger.verbose(JSON.stringify(items));
+			}
+			return;
+		}
+
 		this.info.messages.place.send(this.instance, {
 			items,
 			instance_id: this.instance.id,
@@ -84,8 +95,8 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin {
 	}
 
 	// combinator signals ---------------------------------------------------------
-	async updateStorageEventHandler(message, event) {
-		if (this.stopping) {
+	async updateStorageEventHandler(message) {
+		if (this.instance.status !== "running") {
 			return;
 		}
 		let items = message.data.items;

plugins/subspace_storage/master.js

@@ -5,6 +5,7 @@ const path = require("path");
 const libDatabase = require("@clusterio/lib/database");
 const libPlugin = require("@clusterio/lib/plugin");
 const { Counter, Gauge } = require("@clusterio/lib/prometheus");
+const RateLimiter = require("@clusterio/lib/RateLimiter");
 
 const routes = require("./routes");
 const dole = require("./dole");
@@ -58,6 +59,16 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin {
 	async init() {
 
 		this.items = await loadDatabase(this.master.config, this.logger);
+		this.itemUpdateRateLimiter = new RateLimiter({
+			maxRate: 1,
+			action: () => {
+				try {
+					this.broadcastStorage();
+				} catch (err) {
+					this.logger.error(`Unexpected error sending storage update:\n${err.stack}`);
+				}
+			},
+		});
 		this.itemsLastUpdate = new Map(this.items._items.entries());
 		this.autosaveId = setInterval(() => {
 			saveDatabase(this.master.config, this.items, this.logger).catch(err => {
@@ -78,6 +89,10 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin {
 	}
 
 	updateStorage() {
+		this.itemUpdateRateLimiter.activate();
+	}
+
+	broadcastStorage() {
 		let itemsToUpdate = new Map();
 		for (let [name, count] of this.items._items) {
 			if (this.itemsLastUpdate.get(name) === count) {
@@ -209,6 +224,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin {
 	}
 
 	async onShutdown() {
+		this.itemUpdateRateLimiter.cancel();
 		clearInterval(this.autosaveId);
 		clearInterval(this.doleMagicId);
 		await saveDatabase(this.master.config, this.items, this.logger);

test/integration/clusterio.js

@@ -62,6 +62,42 @@ describe("Integration of Clusterio", function() {
 				await connectorB.close(1000, "");
 			});
 		});
+
+		describe("queryLogRequestHandler", function() {
+			it("should honnor the limit", async function() {
+				let result = await libLink.messages.queryLog.send(getControl(), {
+					all: true,
+					master: false,
+					slave_ids: [],
+					instance_ids: [],
+					max_level: null,
+					limit: 10,
+					order: "asc",
+				});
+				assert.equal(result.log.length, 10);
+			});
+			it("should return entries by order", async function() {
+				let first = await libLink.messages.queryLog.send(getControl(), {
+					all: true,
+					master: false,
+					slave_ids: [],
+					instance_ids: [],
+					max_level: null,
+					limit: 1,
+					order: "asc",
+				});
+				let last = await libLink.messages.queryLog.send(getControl(), {
+					all: true,
+					master: false,
+					slave_ids: [],
+					instance_ids: [],
+					max_level: null,
+					limit: 1,
+					order: "desc",
+				});
+				assert(first.log[0].timestamp < last.log[0].timestamp, "first log entry happened after last");
+			});
+		});
 	});
 
 	describe("clusterioslave", function() {
@@ -88,6 +124,9 @@ describe("Integration of Clusterio", function() {
 					slaveProcess = await spawn(
 						"alt-slave:", `node ../../packages/slave run --config ${config}`, /Started slave/
 					);
+					// Add instance to test the unknown status afterwards
+					await execCtl("instance create alt-test --id 99");
+					await execCtl("instance assign alt-test 5");
 				} finally {
 					if (slaveProcess) {
 						slaveProcess.kill("SIGINT");
@@ -524,7 +563,7 @@ describe("Integration of Clusterio", function() {
 		describe("instanceUpdateEventHandler()", function() {
 			it("should have triggered for the previous instance status updates", function() {
 				let statusesToCheck = new Set([
-					"unassigned", "stopped", "creating_save", "exporting_data",
+					"unassigned", "unknown", "stopped", "creating_save", "exporting_data",
 					"starting", "running", "stopping", "deleted",
 				]);
 				let statusesNotSeen = new Set(statusesToCheck);

test/lib/ExponentialBackoff.js

@@ -4,7 +4,7 @@ const assert = require("assert").strict;
 const ExponentialBackoff = require("@clusterio/lib/ExponentialBackoff");
 
 describe("lib/ExponentialBackoff", function() {
-	describe("class RateLimiter", function() {
+	describe("class ExponentialBackoff", function() {
 		it("should give a delay", function() {
 			let backoff = new ExponentialBackoff();
 			assert.equal(typeof backoff.delay(), "number");