Rework instance assigment process

Under the current system there's no distinction made on the slave side
between instance config files that are present in the filesystem and
instances actually assigned to the slave.  Which could lead to problems
with determing if an instance is actually assigned to the slave.

Resolve this issue by separating discovered instances with assigned
instances.  Upon startup discovered instances are put into the
discoveredInstanceInfos mapping on the Slave object and all of these are
sent to the master server.  The master server in turn sends assignment
requests for all the instances which are assigned to the slave.  These
are put into the instanceInfos mapping of the Slave object which can
then be relied up to include only assigned instances.

packages/ctl/ctl.js

@@ -203,13 +203,13 @@ instanceConfigCommands.add(new libCommand.Command({
 instanceCommands.add(instanceConfigCommands);
 
 instanceCommands.add(new libCommand.Command({
-	definition: ["assign <instance> <slave>", "Assign instance to a slave", (yargs) => {
+	definition: ["assign <instance> [slave]", "Assign instance to a slave", (yargs) => {
 		yargs.positional("instance", { describe: "Instance to assign", type: "string" });
-		yargs.positional("slave", { describe: "Slave to assign to", type: "string" });
+		yargs.positional("slave", { describe: "Slave to assign to or unassign if none", type: "string" });
 	}],
 	handler: async function(args, control) {
 		let instanceId = await libCommand.resolveInstance(control, args.instance);
-		let slaveId = await libCommand.resolveSlave(control, args.slave);
+		let slaveId = args.slave ? await libCommand.resolveSlave(control, args.slave) : null;
 		await libLink.messages.assignInstanceCommand.send(control, {
 			instance_id: instanceId,
 			slave_id: slaveId,

packages/lib/link/messages.js

@@ -411,19 +411,10 @@ messages.assignInstanceCommand = new Request({
 	permission: "core.instance.assign",
 	requestProperties: {
 		"instance_id": { type: "integer" },
-		"slave_id": { type: "integer" },
+		"slave_id": { type: ["integer", "null"] },
 	},
 });
 
-messages.assignInstance = new Request({
-	type: "assign_instance",
-	links: ["master-slave"],
-	requestProperties: {
-		"serialized_config": { type: "object" },
-	},
-	forwardTo: "instance",
-});
-
 messages.startInstance = new Request({
 	type: "start_instance",
 	links: ["control-master", "master-slave", "slave-instance"],
@@ -655,6 +646,44 @@ messages.deleteUser = new Request({
 	},
 });
 
+
+// Internal requests
+messages.updateInstances = new Request({
+	type: "update_instances",
+	links: ["slave-master"],
+	requestProperties: {
+		"instances": {
+			type: "array",
+			items: {
+				type: "object",
+				additionalProperties: false,
+				required: ["serialized_config", "status"],
+				properties: {
+					"serialized_config": { type: "object" },
+					"status": { enum: [
+						"stopped", "starting", "running", "creating_save", "exporting_data",
+					]},
+				},
+			},
+		},
+	},
+});
+
+messages.assignInstance = new Request({
+	type: "assign_instance",
+	links: ["master-slave"],
+	requestProperties: {
+		"serialized_config": { type: "object" },
+	},
+	forwardTo: "instance",
+});
+
+messages.unassignInstance = new Request({
+	type: "unassigne_instance",
+	links: ["master-slave"],
+	forwardTo: "instance",
+});
+
 messages.getMetrics = new Request({
 	type: "get_metrics",
 	links: ["master-slave", "slave-instance"],
@@ -846,27 +875,6 @@ messages.instanceStatusChanged = new Event({
 	},
 });
 
-messages.updateInstances = new Event({
-	type: "update_instances",
-	links: ["slave-master"],
-	eventProperties: {
-		"instances": {
-			type: "array",
-			items: {
-				type: "object",
-				additionalProperties: false,
-				required: ["serialized_config", "status"],
-				properties: {
-					"serialized_config": { type: "object" },
-					"status": { enum: [
-						"stopped", "starting", "running", "creating_save", "exporting_data",
-					]},
-				},
-			},
-		},
-	},
-});
-
 messages.instanceOutput = new Event({
 	type: "instance_output",
 	links: ["instance-slave", "slave-master", "master-control"],

packages/master/master.js

@@ -717,29 +717,43 @@ class ControlConnection extends BaseConnection {
 	}
 
 	async assignInstanceCommandRequestHandler(message, request) {
-		let instance = db.instances.get(message.data.instance_id);
+		let { slave_id, instance_id } = message.data;
+		let instance = db.instances.get(instance_id);
 		if (!instance) {
-			throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`);
+			throw new libErrors.RequestError(`Instance with ID ${instance_id} does not exist`);
 		}
 
-		// XXX: Should the instance be stopped if it's running on another slave?
-
-		let connection = slaveConnections.get(message.data.slave_id);
-		if (!connection) {
-			// The case of the slave not getting the assign instance message
-			// still have to be handled, so it's not a requirement that the
-			// target slave be connected to the master while doing the
-			// assignment, but it is IMHO a better user experience if this
-			// is the case.
-			throw new libErrors.RequestError("Target slave is not connected to the master server");
+		// Check if target slave is connected
+		let newSlaveConnection;
+		if (slave_id !== null) {
+			newSlaveConnection = slaveConnections.get(slave_id);
+			if (!newSlaveConnection) {
+				// The case of the slave not getting the assign instance message
+				// still have to be handled, so it's not a requirement that the
+				// target slave be connected to the master while doing the
+				// assignment, but it is IMHO a better user experience if this
+				// is the case.
+				throw new libErrors.RequestError("Target slave is not connected to the master server");
+			}
 		}
 
-		instance.config.set("instance.assigned_slave", message.data.slave_id);
+		// Unassign from currently assigned slave if it is connected.
+		let currentAssignedSlave = instance.config.get("instance.assigned_slave");
+		if (currentAssignedSlave !== null && slave_id !== currentAssignedSlave) {
+			let oldSlaveConnection = slaveConnections.get(currentAssignedSlave);
+			if (oldSlaveConnection && !oldSlaveConnection.connector.closing) {
+				await libLink.messages.unassignInstance.send(oldSlaveConnection, { instance_id });
+			}
+		}
 
-		return await libLink.messages.assignInstance.send(connection, {
-			instance_id: instance.config.get("instance.id"),
-			serialized_config: instance.config.serialize(),
-		});
+		// Assign to target
+		instance.config.set("instance.assigned_slave", slave_id);
+		if (slave_id !== null) {
+			await libLink.messages.assignInstance.send(newSlaveConnection, {
+				instance_id,
+				serialized_config: instance.config.serialize(),
+			});
+		}
 	}
 
 	async setInstanceOutputSubscriptionsRequestHandler(message) {
@@ -972,13 +986,23 @@ class SlaveConnection extends BaseConnection {
 
 	async instanceStatusChangedEventHandler(message, event) {
 		let instance = db.instances.get(message.data.instance_id);
+
+		// It's possible to get updates from an instance that does not exist
+		// or is not assigned to the slave it originated from if it was
+		// reassigned or deleted while the connection to the slave it was
+		// originally on was down at the time.
+		if (!instance || instance.config.get("instance.assigned_slave") !== this._id) {
+			console.log(`Clusterio | Got bogus update for instance id ${message.data.instance_id}`);
+			return;
+		}
+
 		let prev = instance.status;
 		instance.status = message.data.status;
 		console.log(`Clusterio | Instance ${instance.config.get("instance.name")} State: ${instance.status}`);
 		await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev);
 	}
 
-	async updateInstancesEventHandler(message) {
+	async updateInstancesRequestHandler(message) {
 		// Push updated instance configs
 		for (let instance of db.instances.values()) {
 			if (instance.config.get("instance.assigned_slave") === this._id) {
@@ -996,10 +1020,21 @@ class SlaveConnection extends BaseConnection {
 
 			let masterInstance = db.instances.get(instanceConfig.get("instance.id"));
 			if (masterInstance) {
+				// Check if this instance is assigned somewhere else.
+				if (masterInstance.config.get("instance.assigned_slave") !== this._id) {
+					await libLink.messages.unassignInstance.send(this, {
+						instance_id: masterInstance.config.get("instance.id"),
+					});
+					continue;
+				}
+
 				// Already have this instance, update state instead
 				if (masterInstance.status !== instance.status) {
 					let prev = masterInstance.status;
 					masterInstance.status = instance.status;
+					console.log(
+						`Clusterio | Instance ${instanceConfig.get("instance.name")} State: ${instance.status}`
+					);
 					await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev);
 				}
 				continue;

packages/slave/slave.js

@@ -755,15 +755,16 @@ class Instance extends libLink.Link{
  * Searches for instances in the provided directory
  *
  * Looks through all sub-dirs of the provided directory for valid
- * instance definitions and updates the mapping of instance id to
+ * instance definitions and return a mapping of instance id to
  * instanceInfo objects.
  *
- * @param {Map<integer, Object>} instanceInfos -
- *     mapping between instance id and information about this instance.
  * @param {string} instancesDir - Directory containing instances
  * @param {Object} logger - console like logging interface.
+ * @returns {Map<integer, Object>}
+ *     mapping between instance id and information about this instance.
  */
-async function discoverInstances(instanceInfos, instancesDir, logger) {
+async function discoverInstances(instancesDir, logger) {
+	let instanceInfos = new Map();
 	for (let entry of await fs.readdir(instancesDir, { withFileTypes: true })) {
 		if (entry.isDirectory()) {
 			let instanceConfig = new libConfig.InstanceConfig();
@@ -781,11 +782,6 @@ async function discoverInstances(instanceInfos, instancesDir, logger) {
 				continue;
 			}
 
-			// Ignore instances we have already discovered before.
-			if (instanceInfos.has(instanceConfig.get("instance.id"))) {
-				continue;
-			}
-
 			let instancePath = path.join(instancesDir, entry.name);
 			logger.log(`found instance ${instanceConfig.get("instance.name")} in ${instancePath}`);
 			instanceInfos.set(instanceConfig.get("instance.id"), {
@@ -794,6 +790,7 @@ async function discoverInstances(instanceInfos, instancesDir, logger) {
 			});
 		}
 	}
+	return instanceInfos;
 }
 
 class InstanceConnection extends libLink.Link {
@@ -915,6 +912,7 @@ class Slave extends libLink.Link {
 		this.config = slaveConfig;
 
 		this.instanceConnections = new Map();
+		this.discoveredInstanceInfos = new Map();
 		this.instanceInfos = new Map();
 
 		this.adminlist = new Set();
@@ -1124,20 +1122,28 @@ class Slave extends libLink.Link {
 			console.log(`Clusterio | Updated config for ${instanceInfo.path}`);
 
 		} else {
-			let instanceConfig = new libConfig.InstanceConfig();
-			await instanceConfig.load(serialized_config);
+			instanceInfo = this.discoveredInstanceInfos.get(instance_id);
+			if (instanceInfo) {
+				instanceInfo.config.update(serialized_config, true);
 
-			// XXX: race condition on multiple simultanious calls
-			let instanceDir = await this._findNewInstanceDir(instanceConfig.get("instance.name"));
+			} else {
+				let instanceConfig = new libConfig.InstanceConfig();
+				await instanceConfig.load(serialized_config);
+
+				// XXX: race condition on multiple simultanious calls
+				let instanceDir = await this._findNewInstanceDir(instanceConfig.get("instance.name"));
+
+				await Instance.create(instanceDir, this.config.get("slave.factorio_directory"));
+				instanceInfo = {
+					path: instanceDir,
+					config: instanceConfig,
+				};
+
+				this.discoveredInstanceInfos.set(instance_id, instanceInfo);
+			}
 
-			await Instance.create(instanceDir, this.config.get("slave.factorio_directory"));
-			instanceInfo = {
-				path: instanceDir,
-				config: instanceConfig,
-			};
 			this.instanceInfos.set(instance_id, instanceInfo);
-			console.log(`Clusterio | assigned instance ${instanceConfig.get("instance.name")}`);
-
+			console.log(`Clusterio | assigned instance ${instanceInfo.config.get("instance.name")}`);
 		}
 
 		// Somewhat hacky, but in the event of a lost session the status is
@@ -1160,6 +1166,20 @@ class Slave extends libLink.Link {
 		);
 	}
 
+	async unassignInstanceRequestHandler(message) {
+		let instanceId = message.data.instance_id;
+		let instanceInfo = this.instanceInfos.get(instanceId);
+		if (instanceInfo) {
+			let instanceConnection = this.instanceConnections.get(instanceId);
+			if (instanceConnection && ["starting", "running"].includes(instanceConnection.status)) {
+				await libLink.messages.stopInstance.send(instanceConnection, { instance_id: instanceId });
+			}
+
+			this.instanceInfos.delete(instanceId);
+			console.log(`Clusterio | unassigned instance ${instanceInfo.config.get("instance.name")}`);
+		}
+	}
+
 	/**
 	 * Initialize and connect an unloaded instance
 	 *
@@ -1249,13 +1269,14 @@ class Slave extends libLink.Link {
 			throw new libErrors.RequestError(`Instance with ID ${instanceId} is running`);
 		}
 
-		let instanceInfo = this.instanceInfos.get(instanceId);
+		let instanceInfo = this.discoveredInstanceInfos.get(instanceId);
 		if (!instanceInfo) {
 			throw new libErrors.RequestError(`Instance with ID ${instanceId} does not exist`);
 		}
 
-		await fs.remove(instanceInfo.path);
+		this.discoveredInstanceInfos.delete(instanceId);
 		this.instanceInfos.delete(instanceId);
+		await fs.remove(instanceInfo.path);
 	}
 
 	/**
@@ -1265,16 +1286,16 @@ class Slave extends libLink.Link {
 	 * the slave and master server with the new list of instances.
 	 */
 	async updateInstances() {
-		await discoverInstances(this.instanceInfos, this.config.get("slave.instances_directory"), console);
+		this.discoveredInstanceInfos = await discoverInstances(this.config.get("slave.instances_directory"), console);
 		let list = [];
-		for (let [instanceId, instanceInfo] of this.instanceInfos) {
+		for (let [instanceId, instanceInfo] of this.discoveredInstanceInfos) {
 			let instanceConnection = this.instanceConnections.get(instanceId);
 			list.push({
 				serialized_config: instanceInfo.config.serialize(),
 				status: instanceConnection ? instanceConnection.status : "stopped",
 			});
 		}
-		libLink.messages.updateInstances.send(this, { instances: list });
+		await libLink.messages.updateInstances.send(this, { instances: list });
 	}
 
 	async prepareDisconnectRequestHandler(message, request) {

test/slave.js

@@ -152,8 +152,7 @@ describe("Slave testing", function() {
 		it("should discover test instance", async function() {
 			let logger = { log: () => {}, error: () => {} };
 			let instancePath = path.join("test", "file", "instances");
-			let instanceInfos = new Map();
-			await slave._discoverInstances(instanceInfos, instancePath, logger);
+			let instanceInfos = await slave._discoverInstances(instancePath, logger);
 
 			let referenceConfig = new libConfig.InstanceConfig();
 			await referenceConfig.init();