diff --git a/plugins/inventory_sync/README.md b/plugins/inventory_sync/README.md index c39a8283..ecf5d615 100644 --- a/plugins/inventory_sync/README.md +++ b/plugins/inventory_sync/README.md @@ -32,8 +32,8 @@ The greatly simplified data flow is as follows: 3. If the player inventory was previously uploaded but is not the same then the player inventory is deleted and the player is turned into a spectator. 4. Otherwise the player inventory is kept. -5. Scenario script asks for the player inventory from the controller. -6. Controller sends it to the scenario in a stream of chunks. +5. Scenario script asks for the player inventory from the controller, including its current recipe notification state. +6. Controller sends it to the scenario in a stream of chunks, with only the recipe notifications that differ from the current state. 7. Scenario displays and updates a progress bar as each chunk is received. 8. Once all chunks have loaded the player's synced inventory is recreated from the data and the player can start playing. diff --git a/plugins/inventory_sync/instance.ts b/plugins/inventory_sync/instance.ts index bc05d896..831c7524 100644 --- a/plugins/inventory_sync/instance.ts +++ b/plugins/inventory_sync/instance.ts @@ -1,3 +1,5 @@ +import util from "util"; +import zlib from "zlib"; import * as lib from "@clusterio/lib"; import { BaseInstancePlugin } from "@clusterio/host"; import { @@ -8,6 +10,11 @@ type IpcPlayerName = { player_name: string } +type IpcDownloadRequest = { + player_name: string, + recipe_notifications?: string, +} + type IpcAcquireResponse = { player_name: string, status: string, @@ -26,6 +33,49 @@ function chunkify(chunkSize: number, string: string): string[] { return string.match(new RegExp(`.{1,${chunkSize}}`, "g")) || []; } +const inflate = util.promisify(zlib.inflate); +const deflate = util.promisify(zlib.deflate); + +/** + * Decode a string produced by helpers.encode_string in Factorio + */ +async function decodeLuaString(encoded: string): Promise { + return (await inflate(Buffer.from(encoded, "base64"))).toString("utf8"); +} + +/** + * Encode a string so that helpers.decode_string in Factorio can read it + */ +async function encodeLuaString(text: string): Promise { + return (await deflate(Buffer.from(text, "utf8"))).toString("base64"); +} + +export type RecipeNotificationDelta = { + add?: string[], + remove?: string[], +} + +/** + * Compute the changes needed to turn the current cleared recipe list into the stored one + * @param stored - Cleared recipe names stored on the controller + * @param current - Cleared recipe names the instance currently has + * @returns names to add to and remove from the current list + */ +export function recipeNotificationDelta(stored: string[], current: string[]): RecipeNotificationDelta { + const storedSet = new Set(stored); + const currentSet = new Set(current); + const delta: RecipeNotificationDelta = {}; + const add = stored.filter(name => !currentSet.has(name)); + const remove = current.filter(name => !storedSet.has(name)); + if (add.length) { + delta.add = add; + } + if (remove.length) { + delta.remove = remove; + } + return delta; +} + export class InstancePlugin extends BaseInstancePlugin { playersToRelease!: Set; disconnecting!: boolean; @@ -55,7 +105,7 @@ export class InstancePlugin extends BaseInstancePlugin { ); this.instance.server.on( "ipc-inventory_sync_download", - (request: IpcPlayerName) => this.handleDownload(request).catch( + (request: IpcDownloadRequest) => this.handleDownload(request).catch( err => this.logger.error(`Error handling ipc-inventory_sync_download:\n${err.stack}`) ), ); @@ -163,7 +213,27 @@ export class InstancePlugin extends BaseInstancePlugin { ); } - async handleDownload(request: IpcPlayerName) { + /** + * Replace the stored recipe notifications with the difference from the + * current state on the instance to reduce the amount of data sent to Lua. + */ + async applyRecipeNotificationDelta(playerData: IpcPlayerData, current?: string) { + if (!playerData.recipe_notifications) { + return; + } + try { + const delta = recipeNotificationDelta( + JSON.parse(await decodeLuaString(playerData.recipe_notifications)), + current ? JSON.parse(await decodeLuaString(current)) : [], + ); + playerData.recipe_notifications = await encodeLuaString(JSON.stringify(delta)); + } catch (err: any) { + this.logger.warn(`Dropping invalid recipe notifications for ${playerData.name}:\n${err.stack}`); + delete playerData.recipe_notifications; + } + } + + async handleDownload(request: IpcDownloadRequest) { const playerName = request.player_name; this.logger.verbose(`Downloading ${playerName}`); @@ -177,6 +247,8 @@ export class InstancePlugin extends BaseInstancePlugin { return; } + await this.applyRecipeNotificationDelta(response.playerData, request.recipe_notifications); + const chunkSize = this.instance.config.get("inventory_sync.rcon_chunk_size"); const chunks = chunkify(chunkSize, JSON.stringify(response.playerData)); this.logger.verbose(`Sending inventory for ${playerName} in ${chunks.length} chunks`); diff --git a/plugins/inventory_sync/module/inventory_sync.lua b/plugins/inventory_sync/module/inventory_sync.lua index bf56a57e..c6804251 100644 --- a/plugins/inventory_sync/module/inventory_sync.lua +++ b/plugins/inventory_sync/module/inventory_sync.lua @@ -17,6 +17,7 @@ local progress_dialog = require("modules/inventory_sync/gui/progress_dialog") local dialog_failed_download = require("modules/inventory_sync/gui/dialog_failed_download") local v2_remote_controller = compat.version_ge("2.0.0") +local recipe_notifications_api = compat.version_ge("2.0.67") -- Returns true if the player is currently in a cutscene local function is_in_cutscene(player) @@ -441,8 +442,15 @@ function inventory_sync.initiate_inventory_download(player, player_record, gener local script_data = get_script_data() script_data.active_downloads[player.name] = record + -- The plugin only sends back what differs from the current notification state + local recipe_notifications + if recipe_notifications_api then + recipe_notifications = serialize.serialize_crafting_notifications(player) + end + clusterio_api.send_json("inventory_sync_download", { - player_name = player.name + player_name = player.name, + recipe_notifications = recipe_notifications, }) -- If this is a synced player turn them into a spectator while the diff --git a/plugins/inventory_sync/module/serialize.lua b/plugins/inventory_sync/module/serialize.lua index 6d62a6aa..0ef6705e 100644 --- a/plugins/inventory_sync/module/serialize.lua +++ b/plugins/inventory_sync/module/serialize.lua @@ -533,62 +533,55 @@ function serialize.serialize_crafting_notifications(player, failed) recipes[recipe.name] = nil end - -- Convert to a list of names - local recipe_names = failed or {} -- Includes previous recipes if any - local index = #recipe_names + 1 + -- Convert to a list of names, including previously failed recipes if any + local recipe_names = {} + for _, name in ipairs(failed or {}) do + recipe_names[#recipe_names + 1] = name + end for name in pairs(recipes) do - recipe_names[index] = name - index = index + 1 + recipe_names[#recipe_names + 1] = name end return helpers.encode_string(helpers.table_to_json(recipe_names)) end +--- @class CraftingNotificationDelta +--- @field add string[]? Recipes cleared elsewhere that still have a notification here +--- @field remove string[]? Recipes with a notification elsewhere that are cleared here + --- @param player LuaPlayer ---- @param serialized string +--- @param serialized string Encoded delta against the state sent with the download request --- @return string[]? failed Recipe names that do not exist function serialize.deserialize_crafting_notifications(player, serialized) - -- Decode the recipe name strings - local recipe_names = helpers.json_to_table(assert(helpers.decode_string(serialized))) - assert(type(recipe_names) == "table", "wrong type decoded from json_to_table") - local clear_notification = player.clear_recipe_notification - local add_notification = player.add_recipe_notification - local notifications = {} - local failed = {} + --- @type CraftingNotificationDelta + local delta = helpers.json_to_table(assert(helpers.decode_string(serialized))) + assert(type(delta) == "table", "wrong type decoded from json_to_table") - -- Assume all recipes need a notification + -- Recipes which can have a notification on this server + local enabled = {} for _, recipe in pairs(player.force.recipes) do if recipe.enabled and not recipe.hidden then - notifications[recipe.name] = true + enabled[recipe.name] = true end end - -- Remove recipes the player has cleared - for _, recipe_name in pairs(recipe_names) do - if notifications[recipe_name] then - notifications[recipe_name] = false + -- Clear notifications the player has cleared elsewhere, this means "add seen notification" + local failed = {} + for _, recipe_name in pairs(delta.add or {}) do + if enabled[recipe_name] then + player.clear_recipe_notification(recipe_name) else - -- Recipe does not exist on this server - failed[#failed+1] = recipe_name + failed[#failed + 1] = recipe_name -- Recipe does not exist on this server end end - -- Clear ones which are ones which are not required - for _, recipe in pairs(player.get_recipe_notifications()) do - if notifications[recipe.name] == false then - clear_notification(recipe.name) - end - notifications[recipe.name] = nil -- Prevents double add below - end - - -- Add any which the player did not have - for recipe_name, should_add in pairs(notifications) do - if should_add then - add_notification(recipe_name) + -- Add notifications the player has elsewhere, this means "remove seen notification" + for _, recipe_name in pairs(delta.remove or {}) do + if enabled[recipe_name] then + player.add_recipe_notification(recipe_name) end end - -- Return failed recipes if any return next(failed) and failed or nil end diff --git a/plugins/inventory_sync/test/instance.js b/plugins/inventory_sync/test/instance.js new file mode 100644 index 00000000..020f5979 --- /dev/null +++ b/plugins/inventory_sync/test/instance.js @@ -0,0 +1,130 @@ +import assert from "node:assert/strict"; +import util from "node:util"; +import zlib from "node:zlib"; +import * as lib from "@clusterio/lib"; + +import * as mock from "../../../test/mock.js"; +import { recipeNotificationDelta, InstancePlugin } from "../dist/node/instance.js"; +import { DownloadResponse } from "../dist/node/messages.js"; +import { plugin as info } from "../dist/node/index.js"; + +const inflate = util.promisify(zlib.inflate); +const deflate = util.promisify(zlib.deflate); + +// Matches helpers.encode_string in Factorio +async function encode(value) { + return (await deflate(Buffer.from(JSON.stringify(value)))).toString("base64"); +} +async function decode(encoded) { + return JSON.parse((await inflate(Buffer.from(encoded, "base64"))).toString()); +} + +describe("inventory_sync", function() { + describe("recipeNotificationDelta()", function() { + it("should be empty when the lists match", function() { + assert.deepEqual(recipeNotificationDelta(["a", "b"], ["b", "a"]), {}); + }); + it("should add stored names missing from current", function() { + assert.deepEqual(recipeNotificationDelta(["a", "b", "c"], ["a"]), { add: ["b", "c"] }); + }); + it("should remove current names missing from stored", function() { + assert.deepEqual(recipeNotificationDelta(["a"], ["a", "b"]), { remove: ["b"] }); + }); + it("should handle both directions at once", function() { + assert.deepEqual(recipeNotificationDelta(["a", "b"], ["b", "c"]), { add: ["a"], remove: ["c"] }); + }); + it("should keep duplicates in stored out of the current list", function() { + assert.deepEqual(recipeNotificationDelta(["a", "a"], ["a"]), {}); + }); + }); + + describe("InstancePlugin.applyRecipeNotificationDelta()", function() { + const warnings = []; + const plugin = { logger: { warn: msg => warnings.push(msg) } }; + const apply = InstancePlugin.prototype.applyRecipeNotificationDelta.bind(plugin); + + it("should replace the stored list with an encoded delta", async function() { + const playerData = { name: "test", recipe_notifications: await encode(["a", "b"]) }; + await apply(playerData, await encode(["b", "c"])); + assert.deepEqual(await decode(playerData.recipe_notifications), { add: ["a"], remove: ["c"] }); + }); + it("should send an empty delta when nothing changed", async function() { + const playerData = { name: "test", recipe_notifications: await encode(["a"]) }; + await apply(playerData, await encode(["a"])); + assert.deepEqual(await decode(playerData.recipe_notifications), {}); + }); + it("should add everything without a current list", async function() { + const playerData = { name: "test", recipe_notifications: await encode(["a"]) }; + await apply(playerData, undefined); + assert.deepEqual(await decode(playerData.recipe_notifications), { add: ["a"] }); + }); + it("should do nothing without stored notifications", async function() { + const playerData = { name: "test" }; + await apply(playerData, await encode(["a"])); + assert.deepEqual(playerData, { name: "test" }); + }); + it("should drop the field on invalid input", async function() { + const playerData = { name: "test", recipe_notifications: "not base64 zlib" }; + await apply(playerData, await encode(["a"])); + assert.equal(playerData.recipe_notifications, undefined); + assert.equal(warnings.length, 1); + }); + }); + + describe("InstancePlugin.handleDownload()", function() { + let instancePlugin; + before(async function() { + try { + lib.registerPluginMessages([info]); + } catch (err) { + // Already registered by the full test suite + } + instancePlugin = await mock.createInstancePlugin(InstancePlugin, info); + instancePlugin.instance.mockConfigEntries.set("inventory_sync.rcon_chunk_size", 100000); + }); + beforeEach(function() { + instancePlugin.instance.server.reset(); + }); + + function respondWith(playerData) { + instancePlugin.instance.connector.once("send", message => { + instancePlugin.instance.connector.emit("message", new lib.MessageResponse( + 1, message.dst, message.src, new DownloadResponse(playerData) + )); + }); + } + function downloadCommand(playerData) { + const json = lib.escapeString(JSON.stringify(playerData)); + return `/sc inventory_sync.download_inventory('test','${json}',1,1)`; + } + + it("should send an empty download when the controller has no data", async function() { + respondWith(undefined); + await instancePlugin.handleDownload({ player_name: "test" }); + assert.deepEqual( + instancePlugin.instance.server.rconCommands, + ["/sc inventory_sync.download_inventory('test',nil,0,0)"] + ); + }); + it("should send everything as a delta when no snapshot was given", async function() { + respondWith({ generation: 1, name: "test", recipe_notifications: await encode(["a"]) }); + await instancePlugin.handleDownload({ player_name: "test" }); + const expected = { generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"] }) }; + assert.deepEqual(instancePlugin.instance.server.rconCommands, [downloadCommand(expected)]); + }); + it("should send a delta when a snapshot was given", async function() { + respondWith({ generation: 1, name: "test", recipe_notifications: await encode(["a", "b"]) }); + await instancePlugin.handleDownload({ player_name: "test", recipe_notifications: await encode(["b"]) }); + const expected = { generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"] }) }; + assert.deepEqual(instancePlugin.instance.server.rconCommands, [downloadCommand(expected)]); + }); + it("should be invoked by the download ipc", async function() { + respondWith(undefined); + instancePlugin.instance.server.emit("ipc-inventory_sync_download", { player_name: "test" }); + for (let i = 0; i < 100 && !instancePlugin.instance.server.rconCommands.length; i++) { + await new Promise(resolve => setImmediate(resolve)); + } + assert.equal(instancePlugin.instance.server.rconCommands.length, 1); + }); + }); +});