Made inventories download in chunks to prevent blocking rcon

plugins/inventory_sync/README.md

@@ -10,3 +10,35 @@ Run the following commands in the folder Clusterio is installed to:
     npx clusteriomaster plugin add @clusterio/plugin-inventory_sync
 
 Substitute clusteriomaster with clusterioslave or clusterioctl if this a dedicate slave or ctl installation respectively.
+
+## Method of operation
+
+This plugin does even based synchronization of inventories. The data flow is as follows:
+
+1. Player joins server, gets put in spectator mode
+
+2. Lua script asks instance for inventory
+
+3. Instance asks master for inventory
+
+4. Master does not find inventory, and responds to instance with undefined
+
+5. Instance sets player to survival
+
+6. Player keeps playing, and eventually leaves. Inventory is serialized and sent to instance through stdout/file
+
+7. Instance sends inventory to master
+
+8. Player joins a different server. Lua sends a request to instance, puts player into spectator mode
+
+9. Instance sends request to master
+
+10. Master responds with json data
+
+11. Instance calls command in scenario script using rcon
+
+12. Once the entire command is transfered, the scenario script clears the old inventory, applies the new inventory from the scenario script and sets the player to survival mode.
+
+Communication between the server and instance goes over stdout or rcon, depending on the size of the data. 
+Communication between the instance and the master server goes over websockets.
+Overall, we are able to achieve a latency between 3 and infinite ticks from server join, depending on the size of the inventory. The major limiter is rcon transfer speeds with larger inventories, especially if they contain blueprints.

plugins/inventory_sync/info.js

@@ -39,6 +39,13 @@ InstanceConfigGroup.define({
 	type: "boolean",
 	initial_value: true,
 });
+InstanceConfigGroup.define({
+	name: "rcon_chunk_size",
+	title: "Rcon inventory chunk size",
+	description: "Divide inventories into multiple chunks before sending with rcon to prevent blocking the pipe",
+	type: "number",
+	initial_value: 1000,
+});
 InstanceConfigGroup.finalize();
 
 libUsers.definePermission({

plugins/inventory_sync/instance.js

@@ -8,36 +8,14 @@ const libLuaTools = require("@clusterio/lib/lua_tools");
 class InstancePlugin extends libPlugin.BaseInstancePlugin {
 	async init() {
 		this.inventoryQueue = [];
-
-		// setInterval(() => {
-
-		// 	console.log("Connected")
-		// 	this.info.messages.upload.send(this.instance, {
-		// 		instance_id: this.instance.id,
-		// 		player: "danielv",
-		// 		inventory: "stuff",
-		// 	})
-		// },5000)
-
-
-		this.instance.server.on("ipc-inventory_sync_upload", content => this.handleUpload(content).catch(err => this.logger.error(
-			`Error handling ipc-inventory_sync_upload:\n${err.stack}`
-		))
-		);
-		this.instance.server.on("ipc-inventory_sync_download", content => this.handleDownload(content).catch(err => this.logger.error(
-			`Error handling ipc-inventory_sync_download:\n${err.stack}`
-		))
-		);
-	}
-
-	onMasterConnectionEvent(event) {
-		console.log("Connectionevent", event);
-		if (event === "connect") {
-		}
+		this.instance.server.on("ipc-inventory_sync_upload", content => this.handleUpload(content)
+			.catch(err => this.logger.error(`Error handling ipc-inventory_sync_upload:\n${err.stack}`)));
+		this.instance.server.on("ipc-inventory_sync_download", content => this.handleDownload(content)
+			.catch(err => this.logger.error(`Error handling ipc-inventory_sync_download:\n${err.stack}`)));
 	}
 
 	async handleUpload(player) {
-		console.log("Uploading", player);
+		this.logger.verbose(`Uploading ${player.name} (${JSON.stringify(player).length / 1000}kB)`);
 		this.info.messages.upload.send(this.instance, {
 			instance_id: this.instance.id,
 			instance_name: this.instance.name,
@@ -47,39 +25,29 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin {
 	}
 
 	async handleDownload(player) {
-		console.log("Downloading", player.player_name);
+		this.logger.verbose(`Downloading ${player.player_name}`);
 		let response = await this.info.messages.download.send(this.instance, {
 			player_name: player.player_name,
 		});
 		if (response.inventory) {
-			console.log("Sending command");
-			await this.sendRcon(`/sc inventory_sync.downloadInventory('${response.player_name}', '${libLuaTools.escapeString(response.inventory)}')`);
-		}
-	}
-
-	async chatEventHandler(message) {
-		// TODO check if cross server chat is enabled
-		let content = `[${message.data.instance_name}] ${removeTags(message.data.content)}`;
-		await this.sendRcon(`/sc game.print('${libLuaTools.escapeString(content)}')`, true);
-	}
-
-	sendChat(message) {
-		this.info.messages.chat.send(this.instance, {
-			instance_name: this.instance.name,
-			content: message,
-		});
-	}
-
-	async onOutput(output) {
-		if (output.type === "action" && output.action === "INVENTORY") {
-			if (this.slave.connector.connected) {
-				this.sendChat(output.message);
-			} else {
-				this.inventoryQueue.push(output.message);
+			const chunkSize = this.instance.config.get("inventory_sync.rcon_chunk_size");
+			const chunks = chunkify(chunkSize, response.inventory)
+			this.logger.verbose(`Sending inventory for ${player.player_name} in ${chunks.length} chunks`);
+			for (let i = 0; i < chunks.length; i++) {
+				// this.logger.verbose(`Sending chunk ${i+1} of ${chunks.length}`)
+				await this.sendRcon(`/sc inventory_sync.downloadInventory('${response.player_name}', '${libLuaTools.escapeString(response.inventory)}', ${i + 1}, ${chunks.length})`);
 			}
 		}
 	}
 }
+function chunkify(chunkSize, string) {
+	if (string.length <= chunkSize) {
+		return [string]
+	} else {
+		let chunk = string.substr(0, chunkSize)
+		return [chunk, ...chunkify(chunkSize, string.replace(chunk, ""))]
+	}
+}
 
 module.exports = {
 	InstancePlugin,

plugins/inventory_sync/module/inventory_sync.lua

@@ -34,24 +34,45 @@ end
 function inventory_sync.initiateInventoryDownload(playerIndex)
     game.players[playerIndex].print("Initiating inventory download...")
     global["inv_sync_download_start_tick "..game.players[playerIndex].name] = game.tick
+    if global.download_cache == nil then
+        global.download_cache = {}
+    end
+    global.download_cache[playerIndex] = ""
     clusterio_api.send_json("inventory_sync_download", {
         player_name = game.players[playerIndex].name
     })
 end
 
 -- This function is called by instance.js as /sc inventory_sync.downloadInventory("danielv", Escaped JSON string) with data from the master
-function inventory_sync.downloadInventory(playerName, data)
-    local serialized_player = game.json_to_table(data)
-    local player = game.players[playerName]
-    -- game.print(serpent.block(serialized_player))
-    for _, inv in pairs(inventories) do
-        local inventory = player.get_inventory(inv)
-        if inventory ~= nil and serialized_player.inventories[tostring(inv)] ~= nil then
-            serialize.deserialize_inventory(inventory, serialized_player.inventories[tostring(inv)])
+function inventory_sync.downloadInventory(playerName, data, number, total)
+    if number then
+        if global.download_cache[playerName] == nil then
+            global.download_cache[playerName] = ""
         end
+        -- Append data to download cache
+        global.download_cache[playerName] = global.download_cache[playerName] .. data
+    end
+
+    if number ~= total then
+        -- Show progress in GUI
+        game.players[playerName].print("Downloaded "..number.."/"..total.." parts")
+
+        -- Request next segment
+        rcon.print("Segment downloaded")
+    else
+        local serialized_player = game.json_to_table(global.download_cache[playerName])
+        global.download_cache[playerName] = nil -- remove data to lower mapsize
+        local player = game.players[playerName]
+        -- game.print(serpent.block(serialized_player))
+        for _, inv in pairs(inventories) do
+            local inventory = player.get_inventory(inv)
+            if inventory ~= nil and serialized_player.inventories[tostring(inv)] ~= nil then
+                serialize.deserialize_inventory(inventory, serialized_player.inventories[tostring(inv)])
+            end
+        end
+        local startTick = global["inv_sync_download_start_tick "..game.players[playerName].name]
+        game.print("Imported inventory for "..playerName.." in "..game.tick - startTick.." ticks")
     end
-    local startTick = global["inv_sync_download_start_tick "..game.players[playerName].name]
-    game.print("Imported inventory for "..playerName.." in "..game.tick - startTick.." ticks")
 end
 
 -- Upload inventory when a player leaves the game. Triggers on restart after crash if player was online during crash.